diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index d8d0c32b..cb39b0f2 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -27,6 +27,15 @@ jobs: pip3 install -r docs/requirements.txt - name: run tests ⚙️ run: python3 -m pytest + - name: run tests in offline mode + if: matrix.python-version == '3.10' + run: | + python3 -m pytest \ + -m "not online" \ + --disable-socket \ + --deselect="tests/doctests/wcs_thredds.txt::wcs_thredds.txt" \ + --deselect="tests/doctests/wfs_MapServerWFSCapabilities.txt::wfs_MapServerWFSCapabilities.txt" \ + --deselect="tests/doctests/wms_geoserver_mass_gis.txt::wms_geoserver_mass_gis.txt" - name: run coveralls ⚙️ run: coveralls - name: build docs 🏗️ diff --git a/requirements-dev.txt b/requirements-dev.txt index 67b1d199..628f4358 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -2,6 +2,7 @@ flake8 pytest pytest-cov +pytest-socket Pillow tox twine diff --git a/tests/test_csw_geonetwork.py b/tests/test_csw_geonetwork.py index a08c33d2..43477787 100644 --- a/tests/test_csw_geonetwork.py +++ b/tests/test_csw_geonetwork.py @@ -19,6 +19,7 @@ def test_csw_geonetwork(): SERVICE_URL3 = 'https://metawal.wallonie.be/geonetwork/srv/eng/csw' +@pytest.mark.online @pytest.mark.skipif(not service_ok(SERVICE_URL3), reason='service is unreachable') @pytest.mark.parametrize("esn_in", ['full', 'summary']) diff --git a/tests/test_ogcapi_connectedsystems_osh.py b/tests/test_ogcapi_connectedsystems_osh.py index e2442844..5b9299f7 100644 --- a/tests/test_ogcapi_connectedsystems_osh.py +++ b/tests/test_ogcapi_connectedsystems_osh.py @@ -16,308 +16,338 @@ from owslib.util import Authentication -class OSHFixtures: - NODE_TEST_OK_URL = 'http://34.67.197.57:8585/sensorhub/test' - # Directs to OSH hosted test server - TEST_URL = 'http://34.67.197.57:8585/sensorhub/api/' - auth = Authentication('auto_test', 'automated_tester24') - sml_headers = {'Content-Type': 'application/sml+json'} - json_headers = {'Content-Type': 'application/json'} - geojson_headers = {'Content-Type': 'application/geo+json'} - omjson_headers = {'Content-Type': 'application/om+json'} - - system_definitions = [ - { +@pytest.fixture(scope="session") +def fixtures(): + class OSHFixtures: + NODE_TEST_OK_URL = 'http://34.67.197.57:8585/sensorhub/test' + # Directs to OSH hosted test server + TEST_URL = 'http://34.67.197.57:8585/sensorhub/api/' + auth = Authentication('auto_test', 'automated_tester24') + sml_headers = {'Content-Type': 'application/sml+json'} + json_headers = {'Content-Type': 'application/json'} + geojson_headers = {'Content-Type': 'application/geo+json'} + omjson_headers = {'Content-Type': 'application/om+json'} + + system_definitions = [ + { + "type": "SimpleProcess", + "uniqueId": "urn:osh:sensor:testsmlsensor:001", + "label": "Test SML Sensor", + "description": "A Sensor created from an SML document", + "definition": "http://www.w3.org/ns/ssn/Sensor" + }, + { + "type": "SimpleProcess", + "uniqueId": "urn:osh:sensor:testsmlsensor:002", + "label": "Test SML Sensor #2", + "description": "A Sensor created from an SML document", + "definition": "http://www.w3.org/ns/ssn/Sensor" + } + ] + + sys_sml_to_update = { "type": "SimpleProcess", "uniqueId": "urn:osh:sensor:testsmlsensor:001", "label": "Test SML Sensor", "description": "A Sensor created from an SML document", "definition": "http://www.w3.org/ns/ssn/Sensor" - }, - { + } + + sys_sml_def = { "type": "SimpleProcess", - "uniqueId": "urn:osh:sensor:testsmlsensor:002", - "label": "Test SML Sensor #2", + "uniqueId": "urn:osh:sensor:testsmlsensor:solo", + "label": "Test SML Sensor - Created on its own", "description": "A Sensor created from an SML document", "definition": "http://www.w3.org/ns/ssn/Sensor" } - ] - - sys_sml_to_update = { - "type": "SimpleProcess", - "uniqueId": "urn:osh:sensor:testsmlsensor:001", - "label": "Test SML Sensor", - "description": "A Sensor created from an SML document", - "definition": "http://www.w3.org/ns/ssn/Sensor" - } - - sys_sml_def = { - "type": "SimpleProcess", - "uniqueId": "urn:osh:sensor:testsmlsensor:solo", - "label": "Test SML Sensor - Created on its own", - "description": "A Sensor created from an SML document", - "definition": "http://www.w3.org/ns/ssn/Sensor" - } - - sml_component = { - "type": "SimpleProcess", - "uniqueId": "urn:osh:sensor:testcomponent:001", - "label": "Test Component", - "description": "Test Component Description", - "definition": "http://www.w3.org/ns/ssn/Sensor" - } - - sml_procedure_test_system = {"type": "SimpleProcess", - "uniqueId": "urn:osh:sensor:testsensorwithcomponents:001", - "label": "Test Process/Datastream Sensor", - "description": "A Sensor created to test procedure/datastream creation", - "definition": "http://www.w3.org/ns/ssn/Sensor"} - sml_procedure = { - "type": "SimpleProcess", - "id": "123456789", - "description": "Test Procedure inserted via OWSLib", - "uniqueId": "urn:osh:sensor:testprocedureows:001", - "label": "Test Procedure - OWSLib", - "definition": "http://www.w3.org/ns/sosa/Procedure" - } - - deployment_definition = { - "type": "Feature", - "properties": { - "featureType": "http://www.w3.org/ns/sosa/Deployment", - "uid": "urn:osh:sensor:testdeployment:001", - "name": "Test Deployment 001", - "description": "A test deployment", - "validTime": ["2024-01-01T00:00:00Z", "2024-12-31T23:59:59Z"] - }, - # "geometry": "POINT(-80.0 35.0)" - } - system_id = 'blid74chqmses' - deployment_expected_id = "vssamsrio5eb2" - weatherstation_id = '0s2lbn2n1bnc8' - datastream_id = 'etbrve0msmrre' - - feature_def = { - "geometry": { - "type": "Point", - "coordinates": [-80.0, 35.0] - }, - "type": "Feature", - "properties": { - "featureType": "http://www.w3.org/ns/sosa/Station", - "uid": "urn:osh:sensor:teststation:001", - "name": "Test Station 001", - "description": "A test station", - "parentSystem@link": {"href": "http://localhost:8585/sensorhub/api/systems/blid74chqmses"}, - "sampledFeature@link": { - "href": "https://data.example.com/link/to/resource", - "rel": "alternate", - "type": "application/json", - "hreflang": "en-US", - "title": "Resource Name", - "uid": "urn:x-org:resourceType:0001", - "rt": "http://www.example.org/uri/of/concept", - "if": "http://www.opengis.net/spec/spec-id/version"} + + sml_component = { + "type": "SimpleProcess", + "uniqueId": "urn:osh:sensor:testcomponent:001", + "label": "Test Component", + "description": "Test Component Description", + "definition": "http://www.w3.org/ns/ssn/Sensor" } - } - - ds_definition = { - "name": "Test Datastream", - "outputName": "Test Output #1", - "schema": { - "obsFormat": "application/swe+json", - "encoding": { - "type": "JSONEncoding", - "vectorAsArrays": False + + sml_procedure_test_system = {"type": "SimpleProcess", + "uniqueId": "urn:osh:sensor:testsensorwithcomponents:001", + "label": "Test Process/Datastream Sensor", + "description": "A Sensor created to test procedure/datastream creation", + "definition": "http://www.w3.org/ns/ssn/Sensor"} + sml_procedure = { + "type": "SimpleProcess", + "id": "123456789", + "description": "Test Procedure inserted via OWSLib", + "uniqueId": "urn:osh:sensor:testprocedureows:001", + "label": "Test Procedure - OWSLib", + "definition": "http://www.w3.org/ns/sosa/Procedure" + } + + deployment_definition = { + "type": "Feature", + "properties": { + "featureType": "http://www.w3.org/ns/sosa/Deployment", + "uid": "urn:osh:sensor:testdeployment:001", + "name": "Test Deployment 001", + "description": "A test deployment", + "validTime": ["2024-01-01T00:00:00Z", "2024-12-31T23:59:59Z"] + }, + # "geometry": "POINT(-80.0 35.0)" + } + system_id = 'blid74chqmses' + deployment_expected_id = "vssamsrio5eb2" + weatherstation_id = '0s2lbn2n1bnc8' + datastream_id = 'etbrve0msmrre' + + feature_def = { + "geometry": { + "type": "Point", + "coordinates": [-80.0, 35.0] }, - "recordSchema": { - "type": "DataRecord", - "label": "Test Datastream Record", - "updatable": False, - "optional": False, - "definition": "http://test.com/Record", - "fields": [ - { - "type": "Time", - "label": "Test Datastream Time", - "updatable": False, - "optional": False, - "definition": "http://test.com/Time", - "name": "timestamp", - "uom": { - "href": "http://test.com/TimeUOM" + "type": "Feature", + "properties": { + "featureType": "http://www.w3.org/ns/sosa/Station", + "uid": "urn:osh:sensor:teststation:001", + "name": "Test Station 001", + "description": "A test station", + "parentSystem@link": {"href": "http://localhost:8585/sensorhub/api/systems/blid74chqmses"}, + "sampledFeature@link": { + "href": "https://data.example.com/link/to/resource", + "rel": "alternate", + "type": "application/json", + "hreflang": "en-US", + "title": "Resource Name", + "uid": "urn:x-org:resourceType:0001", + "rt": "http://www.example.org/uri/of/concept", + "if": "http://www.opengis.net/spec/spec-id/version"} + } + } + + ds_definition = { + "name": "Test Datastream", + "outputName": "Test Output #1", + "schema": { + "obsFormat": "application/swe+json", + "encoding": { + "type": "JSONEncoding", + "vectorAsArrays": False + }, + "recordSchema": { + "type": "DataRecord", + "label": "Test Datastream Record", + "updatable": False, + "optional": False, + "definition": "http://test.com/Record", + "fields": [ + { + "type": "Time", + "label": "Test Datastream Time", + "updatable": False, + "optional": False, + "definition": "http://test.com/Time", + "name": "timestamp", + "uom": { + "href": "http://test.com/TimeUOM" + } + }, + { + "type": "Boolean", + "label": "Test Datastream Boolean", + "updatable": False, + "optional": False, + "definition": "http://test.com/Boolean", + "name": "testboolean" } - }, - { - "type": "Boolean", - "label": "Test Datastream Boolean", - "updatable": False, - "optional": False, - "definition": "http://test.com/Boolean", - "name": "testboolean" - } - ] + ] + } } } - } - - systems_api = Systems(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) - procedure_api = Systems(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) - deployment_api = Deployments(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) - sampling_feature_api = SamplingFeatures(TEST_URL, auth=auth, headers=geojson_headers, - alternate_sampling_feature_url='featuresOfInterest') - properties_api = Properties(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) - datastream_api = Datastreams(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) - observations_api = Observations(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) - control_channels_api = ControlChannels(TEST_URL, auth=auth, - headers={'Content-Type': 'application/json'}) - commands_api = Commands(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) - system_events_api = SystemEvents(TEST_URL, auth=auth, headers=omjson_headers) - system_history_api = SystemHistory(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) - - def update_dsid(self, ds_id): - self.datastream_id = ds_id + systems_api = Systems(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) + procedure_api = Systems(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) + deployment_api = Deployments(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) + sampling_feature_api = SamplingFeatures(TEST_URL, auth=auth, headers=geojson_headers, + alternate_sampling_feature_url='featuresOfInterest') + properties_api = Properties(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) + datastream_api = Datastreams(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) + observations_api = Observations(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) + control_channels_api = ControlChannels(TEST_URL, auth=auth, + headers={'Content-Type': 'application/json'}) + commands_api = Commands(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) + system_events_api = SystemEvents(TEST_URL, auth=auth, headers=omjson_headers) + system_history_api = SystemHistory(TEST_URL, auth=auth, headers={'Content-Type': 'application/json'}) + + def update_dsid(self, ds_id): + self.datastream_id = ds_id + + def create_single_system(self): + sys_api = Systems(self.TEST_URL, auth=self.auth, headers=self.sml_headers) + sys_create_res = sys_api.system_create(json.dumps(self.system_definitions[0])) + sys_id = sys_api.response_headers['Location'].split('/')[-1] + return sys_id + + def create_single_datastream(self, system_id: str): + ds_api = Datastreams(self.TEST_URL, auth=self.auth, headers=self.json_headers) + result = ds_api.datastream_create_in_system(system_id, json.dumps(self.ds_definition)) + ds_id = ds_api.response_headers['Location'].split('/')[-1] + return ds_id + + def delete_all_systems(self): + # delete datastreams first + self.delete_all_datastreams() + self.delete_all_sampling_features() + sys_api = Systems(self.TEST_URL, auth=self.auth, headers=self.sml_headers) + systems = sys_api.systems() + for system in systems['items']: + self.systems_api.system_delete(system['id']) + + def delete_all_datastreams(self): + datastreams = self.datastream_api.datastreams() + for ds in datastreams['items']: + self.datastream_api.datastream_delete(ds['id']) + + def delete_all_sampling_features(self): + sampling_features = self.sampling_feature_api.sampling_features(use_fois=True) + for sf in sampling_features['items']: + self.sampling_feature_api.sampling_feature_delete(sf['id'], use_fois=True) + + yield OSHFixtures() -class TestSystems: - fixtures = OSHFixtures() - def test_system_readonly(self): +class TestSystems: + @pytest.mark.online + def test_system_readonly(self, fixtures): # get all systems - res = self.fixtures.systems_api.systems() + res = fixtures.systems_api.systems() assert len(res['items']) > 0 check_ids = ["0s2lbn2n1bnc8", "94n1f19ld7tlc"] assert [any(sys_id == item['id'] for item in res['items']) for sys_id in check_ids] # get a single system - res = self.fixtures.systems_api.system(check_ids[0]) + res = fixtures.systems_api.system(check_ids[0]) assert res is not None assert res['id'] == check_ids[0] @pytest.mark.skip(reason="Skip transactional test") - def test_system_functions(self): + def test_system_functions(self, fixtures): # insertion of systems - self.fixtures.systems_api.headers = self.fixtures.sml_headers - sys_create_res = self.fixtures.systems_api.system_create(json.dumps(self.fixtures.system_definitions)) + fixtures.systems_api.headers = fixtures.sml_headers + sys_create_res = fixtures.systems_api.system_create(json.dumps(fixtures.system_definitions)) assert sys_create_res is not None # update of system and retrieval - sml_desc_copy = self.fixtures.sys_sml_to_update.copy() + sml_desc_copy = fixtures.sys_sml_to_update.copy() sml_desc_copy['description'] = 'Updated Description' sml_str = json.dumps(sml_desc_copy) - post_systems = self.fixtures.systems_api.system_update('blid74chqmses', sml_str) + post_systems = fixtures.systems_api.system_update('blid74chqmses', sml_str) - check_result = self.fixtures.systems_api.system('blid74chqmses') + check_result = fixtures.systems_api.system('blid74chqmses') assert check_result['properties']['description'] == 'Updated Description' # deletion of system - all_systems = self.fixtures.systems_api.systems() + all_systems = fixtures.systems_api.systems() # clear datastreams - delete_all_datastreams() + fixtures.delete_all_datastreams() for system in all_systems['items']: - res = self.fixtures.systems_api.system_delete(system['id']) + res = fixtures.systems_api.system_delete(system['id']) assert res == {} class TestDeployments: - fixtures = OSHFixtures() - @pytest.mark.skip(reason="Skip transactional test") - def test_deployment_create(self): - res1 = self.fixtures.deployment_api.deployment_create(json.dumps(self.fixtures.deployment_definition)) + def test_deployment_create(self, fixtures): + res1 = fixtures.deployment_api.deployment_create(json.dumps(fixtures.deployment_definition)) assert res1 - res2 = self.fixtures.deployment_api.deployments() - assert self.fixtures.deployment_expected_id in [x['id'] for x in res2['items']] - res3 = self.fixtures.deployment_api.deployment(self.fixtures.deployment_expected_id) + res2 = fixtures.deployment_api.deployments() + assert fixtures.deployment_expected_id in [x['id'] for x in res2['items']] + res3 = fixtures.deployment_api.deployment(fixtures.deployment_expected_id) assert res3['properties']['name'] == 'Test Deployment 001' and res3[ - 'id'] == self.fixtures.deployment_expected_id + 'id'] == fixtures.deployment_expected_id @pytest.mark.skip(reason="Skip transactional test") - def test_deployment_update(self): - self.fixtures.deployment_definition['properties']['description'] = 'Updated Description of Deployment 001' - res = self.fixtures.deployment_api.deployment_update(self.fixtures.deployment_expected_id, - json.dumps(self.fixtures.deployment_definition)) + def test_deployment_update(self, fixtures): + fixtures.deployment_definition['properties']['description'] = 'Updated Description of Deployment 001' + res = fixtures.deployment_api.deployment_update(fixtures.deployment_expected_id, + json.dumps(fixtures.deployment_definition)) assert res is not None @pytest.mark.skip(reason="Skip transactional test") - def test_deployment_delete(self): - res = self.fixtures.deployment_api.deployment_delete(self.fixtures.deployment_expected_id) + def test_deployment_delete(self, fixtures): + res = fixtures.deployment_api.deployment_delete(fixtures.deployment_expected_id) assert res is not None class TestSamplingFeatures: - fixtures = OSHFixtures() - - def test_sampling_features_readonly(self): - all_features = self.fixtures.sampling_feature_api.sampling_features(use_fois=True) + @pytest.mark.online + def test_sampling_features_readonly(self, fixtures): + all_features = fixtures.sampling_feature_api.sampling_features(use_fois=True) assert len(all_features['items']) == 51 feature_id = "c4nce3peo8hvc" - feature = self.fixtures.sampling_feature_api.sampling_feature(feature_id, use_fois=True) + feature = fixtures.sampling_feature_api.sampling_feature(feature_id, use_fois=True) assert feature['id'] == feature_id assert feature['properties']['name'] == 'Station WS013' @pytest.mark.skip(reason="Skip transactional test") - def test_sampling_features_all(self): + def test_sampling_features_all(self, fixtures): # setup - delete_all_systems() - system_id = create_single_system() + fixtures.delete_all_systems() + system_id = fixtures.create_single_system() # create a sampling feature - self.fixtures.sampling_feature_api.headers = self.fixtures.geojson_headers - res = self.fixtures.sampling_feature_api.sampling_feature_create(system_id, - json.dumps(self.fixtures.feature_def), True) - assert self.fixtures.sampling_feature_api.response_headers['Location'] is not None - sampling_feature_id = self.fixtures.sampling_feature_api.response_headers['Location'].split('/')[-1] + fixtures.sampling_feature_api.headers = fixtures.geojson_headers + res = fixtures.sampling_feature_api.sampling_feature_create(system_id, + json.dumps(fixtures.feature_def), True) + assert fixtures.sampling_feature_api.response_headers['Location'] is not None + sampling_feature_id = fixtures.sampling_feature_api.response_headers['Location'].split('/')[-1] # get all sampling features - res = self.fixtures.sampling_feature_api.sampling_features(use_fois=True) + res = fixtures.sampling_feature_api.sampling_features(use_fois=True) assert len(res['items']) > 0 assert any(x['id'] == sampling_feature_id for x in res['items']) # get the sampling feature we created - res = self.fixtures.sampling_feature_api.sampling_feature(sampling_feature_id, use_fois=True) + res = fixtures.sampling_feature_api.sampling_feature(sampling_feature_id, use_fois=True) assert res['properties']['name'] == 'Test Station 001' assert res['properties']['featureType'] == 'http://www.w3.org/ns/sosa/Station' # get sampling features from a system - res = self.fixtures.sampling_feature_api.sampling_features_from_system(system_id, use_fois=True) + res = fixtures.sampling_feature_api.sampling_features_from_system(system_id, use_fois=True) assert len(res['items']) > 0 assert any(x['id'] == sampling_feature_id for x in res['items']) # delete the sampling feature - res = self.fixtures.sampling_feature_api.sampling_feature_delete(sampling_feature_id, use_fois=True) - res = self.fixtures.sampling_feature_api.sampling_features(use_fois=True) + res = fixtures.sampling_feature_api.sampling_feature_delete(sampling_feature_id, use_fois=True) + res = fixtures.sampling_feature_api.sampling_features(use_fois=True) assert res == {'items': []} class TestDatastreams: - fixtures = OSHFixtures() - - def test_datastreams_readonly(self): + @pytest.mark.online + def test_datastreams_readonly(self, fixtures): ds_id = 'kjg2qrcm40rfk' - datastreams = self.fixtures.datastream_api.datastreams() + datastreams = fixtures.datastream_api.datastreams() assert len(datastreams['items']) > 0 assert any(x['id'] == ds_id for x in datastreams['items']) - datastream = self.fixtures.datastream_api.datastream(ds_id) + datastream = fixtures.datastream_api.datastream(ds_id) assert datastream['id'] == ds_id assert datastream['name'] == "Simulated Weather Station Network - weather" @pytest.mark.skip(reason="Skip transactional test") - def test_all_ds_functions(self): + def test_all_ds_functions(self, fixtures): # preflight cleanup - delete_all_systems() + fixtures.delete_all_systems() # setup systems needed - self.fixtures.systems_api.headers = self.fixtures.sml_headers - # systems = self.fixtures.systems_api.system_create(json.dumps(self.fixtures.system_definitions)) - system = create_single_system() + fixtures.systems_api.headers = fixtures.sml_headers + # systems = fixtures.systems_api.system_create(json.dumps(fixtures.system_definitions)) + system = fixtures.create_single_system() # insert a datastream - ds_def_str = json.dumps(self.fixtures.ds_definition) - ds_api = Datastreams(self.fixtures.TEST_URL, auth=self.fixtures.auth, headers=self.fixtures.json_headers) + ds_def_str = json.dumps(fixtures.ds_definition) + ds_api = Datastreams(fixtures.TEST_URL, auth=fixtures.auth, headers=fixtures.json_headers) datastream_create = ds_api.datastream_create_in_system(system, ds_def_str) # get the datastream id from Location header @@ -340,25 +370,24 @@ def test_all_ds_functions(self): class TestObservations: - fixtures = OSHFixtures() - - def test_observations_readonly(self): + @pytest.mark.online + def test_observations_readonly(self, fixtures): ds_id = 'kjg2qrcm40rfk' - observations = self.fixtures.observations_api.observations_of_datastream(ds_id) + observations = fixtures.observations_api.observations_of_datastream(ds_id) assert len(observations['items']) > 0 assert 'result' in observations['items'][0] - observation_of_ds = self.fixtures.observations_api.observations_of_datastream(ds_id) + observation_of_ds = fixtures.observations_api.observations_of_datastream(ds_id) assert observation_of_ds['items'][0]['result']['stationID'] == "WS013" keys = ['stationID', 'temperature', 'pressure', 'humidity', 'windSpeed', 'windDirection'] assert [key in observation_of_ds['items'][0]['result'] for key in keys] @pytest.mark.skip(reason="Skip transactional test") - def test_observations(self): + def test_observations(self, fixtures): # setup - delete_all_systems() - system = create_single_system() - ds = create_single_datastream(system) + fixtures.delete_all_systems() + system = fixtures.create_single_system() + ds = fixtures.create_single_datastream(system) the_time = datetime.utcnow().isoformat() + 'Z' observation = { @@ -369,60 +398,23 @@ def test_observations(self): "testboolean": True } } - self.fixtures.observations_api.headers = {'Content-Type': 'application/om+json'} - res = self.fixtures.observations_api.observations_create_in_datastream(ds, json.dumps(observation)) - obs = self.fixtures.observations_api.observations_of_datastream(ds) + fixtures.observations_api.headers = {'Content-Type': 'application/om+json'} + res = fixtures.observations_api.observations_create_in_datastream(ds, json.dumps(observation)) + obs = fixtures.observations_api.observations_of_datastream(ds) assert obs['items'][0]['phenomenonTime'] == the_time obs_id = obs['items'][0]['id'] - res = self.fixtures.observations_api.observations_delete(obs_id) - obs = self.fixtures.observations_api.observations_of_datastream(ds) + res = fixtures.observations_api.observations_delete(obs_id) + obs = fixtures.observations_api.observations_of_datastream(ds) assert obs['items'] == [] - delete_all_systems() + fixtures.delete_all_systems() class TestSystemHistory: - fixtures = OSHFixtures() - - def test_system_history(self): + @pytest.mark.online + def test_system_history(self, fixtures): sys_id = '0s2lbn2n1bnc8' - res = self.fixtures.system_history_api.system_history(sys_id) + res = fixtures.system_history_api.system_history(sys_id) assert len(res['items']) > 0 history_id = res['items'][0]['properties']['validTime'][0] - res = self.fixtures.system_history_api.system_history_by_id(system_id=sys_id, history_id=history_id) + res = fixtures.system_history_api.system_history_by_id(system_id=sys_id, history_id=history_id) assert res['id'] == sys_id - - -def create_single_system(): - sys_api = Systems(OSHFixtures.TEST_URL, auth=OSHFixtures.auth, headers=OSHFixtures.sml_headers) - sys_create_res = sys_api.system_create(json.dumps(OSHFixtures.system_definitions[0])) - sys_id = sys_api.response_headers['Location'].split('/')[-1] - return sys_id - - -def create_single_datastream(system_id: str): - ds_api = Datastreams(OSHFixtures.TEST_URL, auth=OSHFixtures.auth, headers=OSHFixtures.json_headers) - result = ds_api.datastream_create_in_system(system_id, json.dumps(OSHFixtures.ds_definition)) - ds_id = ds_api.response_headers['Location'].split('/')[-1] - return ds_id - - -def delete_all_systems(): - # delete datastreams first - delete_all_datastreams() - delete_all_sampling_features() - sys_api = Systems(OSHFixtures.TEST_URL, auth=OSHFixtures.auth, headers=OSHFixtures.sml_headers) - systems = sys_api.systems() - for system in systems['items']: - OSHFixtures.systems_api.system_delete(system['id']) - - -def delete_all_datastreams(): - datastreams = OSHFixtures.datastream_api.datastreams() - for ds in datastreams['items']: - OSHFixtures.datastream_api.datastream_delete(ds['id']) - - -def delete_all_sampling_features(): - sampling_features = OSHFixtures.sampling_feature_api.sampling_features(use_fois=True) - for sf in sampling_features['items']: - OSHFixtures.sampling_feature_api.sampling_feature_delete(sf['id'], use_fois=True) diff --git a/tox.ini b/tox.ini index 4163ee91..935ef79e 100644 --- a/tox.ini +++ b/tox.ini @@ -1,5 +1,16 @@ [pytest] -addopts = -v -rxs -s --color=yes --tb=native --ignore=setup.py --doctest-modules --doctest-glob 'tests/**/*.txt' --cov-report term-missing --cov owslib +addopts = + -v + -rxs + -s + --color=yes + --tb=native + --ignore=setup.py + --doctest-modules + --doctest-glob='tests/**/*.txt' + --cov-report=term-missing + --cov=owslib + norecursedirs = .git docs examples etc cov* *.egg* pytest* .tox _broken markers = online: test requires online resources.