import pytest
import json
from .models import StationmetaCore
from toardb.auth_user.models import AuthUser
from toardb.contacts.models import Person, Organisation
from .schemas import get_geom_from_coordinates, Coordinates
from toardb.test_base import (
client,
get_test_db,
create_test_database,
url,
get_test_engine,
test_db_session as db,
)
class TestApps:
def setup(self):
self.application_url = "/stationmeta_core/"
"""Set up all the data before each test
If you want the setup only once (per test module),
the scope argument is not working in the expected way, as discussed here:
https://stackoverflow.com/questions/45817153/py-test-fixture-use-function-fixture-in-scope-fixture
"""
@pytest.fixture(autouse=True)
def setup_db_data(self, db):
# id_seq will not be reset automatically between tests!
_db_conn = get_test_engine()
fake_conn = _db_conn.raw_connection()
fake_cur = fake_conn.cursor()
fake_cur.execute("ALTER SEQUENCE auth_user_id_seq RESTART WITH 1")
fake_conn.commit()
fake_cur.execute("ALTER SEQUENCE stationmeta_core_id_seq RESTART WITH 1")
fake_conn.commit()
fake_cur.execute("ALTER SEQUENCE stationmeta_global_id_seq RESTART WITH 1")
fake_conn.commit()
fake_cur.execute("ALTER SEQUENCE stationmeta_global_services_id_seq RESTART WITH 1")
fake_conn.commit()
fake_cur.execute("ALTER SEQUENCE stationmeta_annotations_id_seq RESTART WITH 1")
fake_conn.commit()
fake_cur.execute("ALTER SEQUENCE stationmeta_roles_id_seq RESTART WITH 1")
fake_conn.commit()
fake_cur.execute("ALTER SEQUENCE stationmeta_aux_doc_id_seq RESTART WITH 1")
fake_conn.commit()
fake_cur.execute("ALTER SEQUENCE stationmeta_aux_image_id_seq RESTART WITH 1")
fake_conn.commit()
fake_cur.execute("ALTER SEQUENCE stationmeta_aux_url_id_seq RESTART WITH 1")
fake_conn.commit()
fake_cur.execute("ALTER SEQUENCE persons_id_seq RESTART WITH 1")
fake_conn.commit()
fake_cur.execute("ALTER SEQUENCE organisations_id_seq RESTART WITH 1")
fake_conn.commit()
infilename = "toardb/auth_user/fixtures/auth.json"
with open(infilename) as f:
metajson=json.load(f)
for entry in metajson:
new_auth_user= AuthUser(**entry)
db.add(new_auth_user)
db.commit()
db.refresh(new_auth_user)
infilename = "toardb/contacts/fixtures/persons.json"
with open(infilename) as f:
metajson=json.load(f)
for entry in metajson:
new_person = Person(**entry)
db.add(new_person)
db.commit()
db.refresh(new_person)
infilename = "toardb/contacts/fixtures/organisations.json"
with open(infilename) as f:
metajson=json.load(f)
for entry in metajson:
new_organisation = Organisation(**entry)
db.add(new_organisation)
db.commit()
db.refresh(new_organisation)
# I also need to upload tests with nested data!!!
infilename = "toardb/stationmeta/fixtures/stationmeta_core.json"
with open(infilename) as f:
metajson=json.load(f)
for entry in metajson:
new_stationmeta_core = StationmetaCore(**entry)
# there's a mismatch with coordinates --> how to automatically switch back and forth?!
tmp_coordinates = new_stationmeta_core.coordinates
new_stationmeta_core.coordinates = get_geom_from_coordinates(Coordinates(**new_stationmeta_core.coordinates))
# there's also a mismatch with additional_metadata --> BUT: this should not be switched back!
# in upload command, we have now: "additional_metadata": "{}"
# but return from this method gives: "additional_metadata": {}
# ==> there is a mismatch between model(JSONB) and schema(JSON)
new_stationmeta_core.additional_metadata = str(new_stationmeta_core.additional_metadata)
db.add(new_stationmeta_core)
db.commit()
db.refresh(new_stationmeta_core)
## def test_get_stationmeta(self, client, db):
## response = client.get("/stationmeta_core/")
## expected_status_code = 200
## assert response.status_code == expected_status_code
## expected_resp = [{'id': 1, 'codes': ['China11'], 'name': 'Mount Tai',
## 'coordinates': {'lat': 36.256, 'lng': 117.106, 'alt': 1534.0},
## 'country': 'China', 'state': 'Shandong Sheng',
## 'coordinate_validation_status': 0, 'coordinate_validation_date': '2020-02-28T12:27:03.746260+01:00',
## 'type_of_environment': 0, 'type_of_area': 0, 'category': '', 'timezone': '',
## 'additional_metadata': {}, 'coordinate_validator_id': 1},
## {'id': 2, 'codes': ['SDZ54421'], 'name': 'Shangdianzi',
## 'coordinates': {'lat': 40.65, 'lng': 117.106, 'alt': 293.9},
## 'country': 'China', 'state': 'Beijing Shi',
## 'coordinate_validation_status': 0, 'coordinate_validation_date': '2020-02-28T12:27:03.746260+01:00',
## 'type_of_environment': 0, 'type_of_area': 0, 'category': '', 'timezone': '',
## 'additional_metadata': {}, 'coordinate_validator_id': 1},
## {'id': 3, 'codes': ['China_test8'], 'name': 'Test_China',
## 'coordinates': {'lat': 36.256, 'lng': 117.106, 'alt': 1534.0},
## 'country': 'China', 'state': 'Shandong Sheng',
## 'coordinate_validation_status': 0, 'coordinate_validation_date': '2020-03-11T12:22:18.047974+01:00',
## 'type_of_environment': 0, 'type_of_area': 0, 'category': '', 'timezone': '',
## 'additional_metadata': {}, 'coordinate_validator_id': 1}]
## assert response.json() == expected_resp
# def test_get_all(self):
# response = client.get("/stationmeta_nested/")
# expected_status_code = 200
# assert response.status_code == expected_status_code
# expected_resp = [OrderedDict([('codes', ['China11']), ('name', 'Mount Tai'), ('country', 'China'),
# ('state', 'Shandong Sheng'), ('coordinates', 'SRID=4326;POINT Z (17.106 36.256 1534)'),
# ('coordinate_validation_status', 'not checked'), ('coordinate_validator', 'sschroeder'),
# ('coordinate_validation_date', '2020-03-11T11:22:18.047000Z'),
# ('roles', [OrderedDict([('rolecode', 'PointOfContact'), ('person', 'Sabine Schröder <s.schroeder@fz-juelich.de>'), ('status', 'active')])]),
# ('meta_global', None), ('additional_metadata', {}),
# ('stationmetaauxurl', []), ('stationmetaauxdoc', []), ('stationmetaauximage', []), ('annotations', [])])]
# assert response.json() == expected_resp
def test_get_special(self, client, db): # core!
response = client.get("/stationmeta_core/China_test8")
expected_status_code = 200
assert response.status_code == expected_status_code
expected_resp = {'id': 3, 'codes': ['China_test8'], 'name': 'Test_China',
'coordinates': {'lat': 36.256, 'lng': 117.106, 'alt': 1534.0},
'country': 'China', 'state': 'Shandong Sheng', 'coordinate_validation_status': 'NotChecked',
'coordinate_validation_date': '2020-03-11T12:22:18.047974+01:00', 'type_of_environment': 'Unknown',
'type_of_area': 'Unknown', 'category': '', 'timezone': '',
'additional_metadata': {}, 'coordinate_validator_id': 1}
assert response.json() == expected_resp
def test_get_special_nested(self, client, db): # nested!
response = client.get("/stationmeta/China_test8")
expected_status_code = 200
assert response.status_code == expected_status_code
expected_resp = {'id': 3, 'codes': ['China_test8'], 'name': 'Test_China',
'coordinates': {'lat': 36.256, 'lng': 117.106, 'alt': 1534.0},
'country': 'China', 'state': 'Shandong Sheng', 'coordinate_validation_status': 'NotChecked',
'coordinate_validation_date': '2020-03-11T12:22:18.047974+01:00', 'type_of_environment': 'Unknown',
'type_of_area': 'Unknown', 'category': '', 'timezone': '',
'additional_metadata': {}, 'coordinate_validator_id': 1,
'roles': [], 'annotations': [], 'aux_images': [], 'aux_docs': [],
'aux_urls': [], 'globalmeta': None, 'globalservice': None}
assert response.json() == expected_resp
# def test_insert_new_without_credits(self):
#? response = client.post("/stationmeta_core/")
# expected_status_code=401
# assert response.status_code == expected_status_code
#? expected_resp = ...
# assert response.json() == expected_resp
# def test_insert_new_wrong_credits(self):
#? response = client.post("/stationmeta_core/")
# expected_status_code = 401
# assert response.status_code == expected_status_code
#? expected_resp = ...
# assert response.json() == expected_resp
def test_insert_new(self, client, db):
# response = client.post("/stationmeta/",
# json={"stationmeta":
# {"codes":["ttt3","ttt4"],
# "name":"Test_China","coordinates":{"lat":36.256,"lng":117.106,"alt":1534.0},
# "country":"China","state":"Shandong Sheng","coordinate_validation_status":"NotChecked",
# "coordinate_validation_date":"2020-03-11T12:22:18.047974+01:00",
# "type_of_environment":"Unknown","type_of_area":"Unknown","category":"","timezone":"",
# "coordinate_validator_id": 1, "additional_metadata":"{}"}
# }
# )
response = client.post("/stationmeta/",
json={"stationmeta":
{"codes":["ttt3","ttt4"],
"name":"Test_China","coordinates":{"lat":36.256,"lng":117.106,"alt":1534.0},
"country":"China","state":"Shandong Sheng","coordinate_validation_status":"NotChecked",
"coordinate_validation_date":"2020-03-11T12:22:18.047974+01:00",
"type_of_environment":"Unknown","type_of_area":"Unknown","category":"","timezone":"",
"coordinate_validator_id": 1, "additional_metadata":"{}"}
}
)
expected_status_code = 200
assert response.status_code == expected_status_code
expected_resp = {'id': 4, 'codes': ['ttt3','ttt4'], 'name': 'Test_China',
'coordinates': {'lat': 36.256, 'lng': 117.106, 'alt': 1534.0},
'country': 'China', 'state': 'Shandong Sheng', 'coordinate_validation_status': 'NotChecked',
'coordinate_validation_date': '2020-03-11T12:22:18.047974+01:00', 'type_of_environment': 'Unknown',
'type_of_area': 'Unknown', 'category': '', 'timezone': '',
'additional_metadata': {}, 'coordinate_validator_id': 1,
'roles': [], 'annotations': [], 'aux_images': [], 'aux_docs': [], 'aux_urls': [],
'globalmeta': None, 'globalservice': None}
assert response.json() == expected_resp
def test_insert_duplicate(self, client, db):
response = client.post("/stationmeta/",
json={"stationmeta":
{"codes":["China11"],
"name":"Test_China","coordinates":{"lat":36.256,"lng":117.106,"alt":1534.0},
"country":"China","state":"Shandong Sheng","coordinate_validation_status":"NotChecked",
"coordinate_validation_date":"2020-03-11T12:22:18.047974+01:00",
"type_of_environment":"Unknown","type_of_area":"Unknown","category":"","timezone":"",
"coordinate_validator_id": 1, "additional_metadata":"{}"}
}
)
expected_status_code = 400
assert response.status_code == expected_status_code
expected_resp = {'detail': 'Station already registered.'}
assert response.json() == expected_resp