250 lines
7.5 KiB
Python
250 lines
7.5 KiB
Python
import os
|
|
|
|
from flask import Flask
|
|
import pytest
|
|
import yaml
|
|
from flask.testing import FlaskClient
|
|
|
|
from app.app import create_app
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_config(tmpdir) -> str:
|
|
configfile = str(tmpdir.join('config.yml'))
|
|
vrd_path = tmpdir.mkdir('vrds')
|
|
dir_path = tmpdir.mkdir('pubs')
|
|
apache_config = tmpdir.join('apache.cfg')
|
|
apache_config.write('#start\n')
|
|
flagfile = tmpdir.join('apache_restart')
|
|
platform_path = tmpdir.mkdir('platform')
|
|
ws_module = 'wsap24.so'
|
|
platform_path.join(ws_module).write('\0')
|
|
|
|
with open(configfile, 'w') as f:
|
|
yaml.dump({
|
|
'apache_config': str(apache_config),
|
|
'vrd_path': str(vrd_path),
|
|
'dir_path': str(dir_path),
|
|
'url_base': '/1c',
|
|
'platform_path': str(platform_path),
|
|
'ws_module': ws_module,
|
|
'vrd_params': {
|
|
'debug': None,
|
|
'server_addr': 'localhost'
|
|
},
|
|
'infobases': {
|
|
'server_file': 'test/1CV8Clst.lst',
|
|
},
|
|
'apache_restart_flagfile': str(flagfile),
|
|
'url_prefix': 'http://localhost',
|
|
}, f)
|
|
return configfile
|
|
|
|
|
|
@pytest.fixture
|
|
def flask_app(temp_config) -> Flask:
|
|
os.environ['WEBPUB1C_CONFIG'] = temp_config
|
|
return create_app()
|
|
|
|
|
|
@pytest.fixture
|
|
def client(flask_app) -> FlaskClient:
|
|
return flask_app.test_client()
|
|
|
|
|
|
def test_index(client):
|
|
response = client.get('/')
|
|
assert b'<!DOCTYPE html>' in response.data
|
|
|
|
|
|
def test_api_index(client):
|
|
response = client.get('/api/v1/')
|
|
data = response.get_json()
|
|
assert data == ['infobases-available', 'infobases-all', 'publications',
|
|
'module', 'config', 'config-test', 'apache-restart']
|
|
|
|
|
|
def test_api_restart_flag(client):
|
|
endpoint = '/api/v1/apache-restart'
|
|
# test when flag not set
|
|
response = client.get(endpoint)
|
|
assert 404 == response.status_code
|
|
# set flag
|
|
response = client.put(endpoint)
|
|
assert 200 == response.status_code
|
|
# test when flag is set
|
|
response = client.get(endpoint)
|
|
assert 200 == response.status_code
|
|
|
|
|
|
def test_api_config_test(client):
|
|
response = client.get('/api/v1/config-test')
|
|
data = response.get_json()
|
|
assert data == {
|
|
'is_apache_cfg_valid': True,
|
|
'is_vrd_path_valid': True,
|
|
'is_dir_path_valid': True,
|
|
'is_url_base_valid': True,
|
|
'is_module_valid': True,
|
|
}
|
|
|
|
|
|
def test_api_config(client):
|
|
response = client.get('/api/v1/config')
|
|
data = response.get_json()
|
|
expected = ['apache_config', 'apache_restart_flagfile', 'infobases', 'dir_path', 'platform_path',
|
|
'url_base', 'url_prefix', 'vrd_params', 'vrd_path', 'ws_module']
|
|
expected.sort()
|
|
actual = list(data.keys())
|
|
actual.sort()
|
|
assert actual == expected
|
|
|
|
|
|
def test_api_module(client):
|
|
endpoint = '/api/v1/module'
|
|
# test when module not added
|
|
response = client.get(endpoint)
|
|
assert 404 == response.status_code
|
|
# add module
|
|
response = client.put(endpoint)
|
|
assert 200 == response.status_code
|
|
# test when module added
|
|
response = client.get(endpoint)
|
|
assert 200 == response.status_code
|
|
# try to add module again
|
|
response = client.put(endpoint)
|
|
assert 304 == response.status_code
|
|
|
|
|
|
def test_api_infobases_available(client):
|
|
response = client.get('/api/v1/infobases-available')
|
|
data = response.get_json()
|
|
expected = ['test1', 'test2', 'bpdemo']
|
|
expected.sort()
|
|
actual = data
|
|
actual.sort()
|
|
assert actual == expected
|
|
|
|
|
|
def test_api_publications(client):
|
|
response = client.get('/api/v1/publications')
|
|
data = response.get_json()
|
|
assert data == []
|
|
|
|
|
|
def test_api_publications_add(client):
|
|
endpoint = '/api/v1/publications'
|
|
# before add
|
|
response = client.get(endpoint)
|
|
data = response.get_json()
|
|
assert data == []
|
|
# add
|
|
response = client.put(path=endpoint, data={'name': 'test123'})
|
|
data = response.get_json()
|
|
assert data['message'] == 'created'
|
|
assert data['name'] == 'test123'
|
|
# add same publication
|
|
response = client.put(path=endpoint, data={'name': 'test123'})
|
|
assert response.status_code == 409
|
|
|
|
|
|
def test_api_publications_add_force(client):
|
|
endpoint = '/api/v1/publications'
|
|
# add
|
|
response = client.put(path=endpoint, data={'name': 'test123'})
|
|
assert response.status_code == 201
|
|
# add same publication
|
|
response = client.put(path=endpoint, data={'name': 'test123'})
|
|
assert response.status_code == 409
|
|
# add same publication with force
|
|
response = client.put(path=endpoint, data={'name': 'test123', 'force': True})
|
|
assert response.status_code == 201
|
|
|
|
|
|
def test_api_publications_add_and_list(client):
|
|
endpoint = '/api/v1/publications'
|
|
# add publications
|
|
client.put(path=endpoint, data={'name': 'test123'})
|
|
client.put(path=endpoint, data={'name': 'test456', 'file': '/path/to/infobase'})
|
|
# get list
|
|
response = client.get(endpoint)
|
|
data = response.get_json()
|
|
assert len(data) == 2
|
|
# test123
|
|
assert data[0]['name'] == 'test123'
|
|
assert data[0]['publicated']
|
|
assert data[0]['url'] == '/1c/test123'
|
|
assert not data[0]['is_file_infobase']
|
|
# test456 / file
|
|
assert data[1]['name'] == 'test456'
|
|
assert data[1]['publicated']
|
|
assert data[1]['url'] == '/1c/test456'
|
|
assert data[1]['is_file_infobase']
|
|
assert data[1]['infobase_filepath'] == '/path/to/infobase'
|
|
|
|
|
|
def test_api_publication_get(client):
|
|
# add publications
|
|
client.put(path='/api/v1/publications', data={'name': 'test123'})
|
|
# get
|
|
response = client.get('/api/v1/publications/test123')
|
|
data = response.get_json()
|
|
assert 'test123' in data
|
|
data = data['test123']
|
|
assert data['name'] == 'test123'
|
|
assert data['publicated']
|
|
assert data['url'] == '/1c/test123'
|
|
assert not data['is_file_infobase']
|
|
|
|
|
|
def test_api_publication_remove(client):
|
|
# add publications
|
|
client.put(path='/api/v1/publications', data={'name': 'test123'})
|
|
# get
|
|
response = client.get('/api/v1/publications/test123')
|
|
assert response.status_code == 200
|
|
# remove
|
|
response = client.delete('/api/v1/publications/test123')
|
|
assert response.status_code == 200
|
|
# get
|
|
response = client.get('/api/v1/publications/test123')
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_api_publication_set_url(client):
|
|
# add publications
|
|
client.put(path='/api/v1/publications', data={'name': 'test123', 'url': 'some/url'})
|
|
# get
|
|
response = client.get('/api/v1/publications/test123/url')
|
|
data = response.get_json()
|
|
assert data['url'] == '/1c/some/url'
|
|
# set url
|
|
response = client.post(path='/api/v1/publications/test123/url', data={'url': '/another/url'})
|
|
assert response.status_code == 200
|
|
# get
|
|
response = client.get('/api/v1/publications/test123/url')
|
|
data = response.get_json()
|
|
assert data['url'] == '/1c/another/url'
|
|
|
|
|
|
def test_api_infobases_all(client):
|
|
# add publications
|
|
client.put(path='/api/v1/publications', data={'name': 'test1'})
|
|
# get list
|
|
response = client.get('/api/v1/infobases-all')
|
|
data = response.get_json()
|
|
assert len(data) == 3
|
|
# bpdemo
|
|
assert data[0]['name'] == 'bpdemo'
|
|
assert not data[0]['publicated']
|
|
assert data[0]['url'] == ''
|
|
# test1
|
|
assert data[1]['name'] == 'test1'
|
|
assert data[1]['publicated']
|
|
assert data[1]['url'] == '/1c/test1'
|
|
assert not data[1]['is_file_infobase']
|
|
# test2
|
|
assert data[2]['name'] == 'test2'
|
|
assert not data[2]['publicated']
|
|
assert data[2]['url'] == ''
|