DRY with conftest

Dockerfile

@@ -4,7 +4,7 @@ ARG CKAN_VERSION
 
 USER root
 
-RUN apt-get update && apt-get install -y postgresql-client openjdk-17-jre geos-dev
+RUN apt-get update && apt-get install -y postgresql-client openjdk-17-jre libgeos-dev
 
 # Download Saxon jar for FGDC2ISO transform (geodatagov)
 ARG saxon_ver=9.9.1-7

ckanext/geodatagov/tests/conftest.py

@@ -0,0 +1,11 @@
+import pytest
+
+import utils
+
+@pytest.fixture(scope="session", autouse=True)
+def run_once_for_all_tests():
+    utils.simple_http_server()
+
+@pytest.fixture(scope="class", autouse=True)
+def run_for_every_test_class():
+    utils.reset_db_and_solr()

ckanext/geodatagov/tests/test_category_tags.py

@@ -1,6 +1,7 @@
 import logging
 import json
 import pytest
+
 from ckan import plugins as p
 from ckan.tests import factories
 

ckanext/geodatagov/tests/test_datajson.py

@@ -1,45 +1,23 @@
 import json
-import os
 import pytest
+import logging
 
-from ckan.tests.helpers import reset_db
 from ckan.tests.factories import Organization
 from ckan import model
-from ckan.model.meta import Session, metadata
 from factories import (DataJsonHarvestSourceObj,
                        HarvestJobObj)
 import ckanext.harvest.model as harvest_model
 from ckanext.datajson.harvester_datajson import DataJsonHarvester
-import mock_static_file_server
-import logging
 
+from utils import PORT, populate_locations_table
 
 log = logging.getLogger(__name__)
 
 @pytest.mark.usefixtures("with_plugins")
 class TestDataJsonHarvester(object):
 
-    @classmethod
-    def setup_class(cls):
-        log.info('Starting mock http server')
-        mock_static_file_server.serve(port=8996)        
-
     def setup_method(self):
-        # https://github.com/ckan/ckan/issues/4764
-        # drop extension postgis so we can reset db
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'drop extension IF EXISTS postgis cascade;'")
-        reset_db()
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'create extension postgis;'")
-        # echo "Downloading locations table"
-        os.system("wget https://github.com/GSA/datagov-deploy/raw/71936f004be1882a506362670b82c710c64ef796/"
-                  "ansible/roles/software/ec2/ansible/files/locations.sql.gz -O /tmp/locations.sql.gz")
-        # echo "Creating locations table"
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'DROP TABLE IF EXISTS locations;'")
-        os.system("gunzip -c /tmp/locations.sql.gz | PGPASSWORD=ckan psql -h db -U ckan -d ckan -v ON_ERROR_STOP=1")
-        # echo "Cleaning"
-        os.system("rm -f /tmp/locations.sql.gz")
-        # os.system("ckan -c test.ini db upgrade -p harvest")
-        metadata.create_all(bind=Session.bind)
+        populate_locations_table()
 
     def run_gather(self, url):
         source = DataJsonHarvestSourceObj(url=url, owner_org=self.organization['id'])
@@ -103,7 +81,7 @@ def test_sample5_data(self):
         self.organization = Organization()
 
         # testing with data from https://www.consumerfinance.gov/data.json
-        url = 'http://127.0.0.1:8996/sample5_data.json'
+        url = f'http://127.0.0.1:{PORT}/sample5_data.json'
         obj_ids = self.run_gather(url=url)
         assert len(obj_ids) == 2
         self.run_fetch()

ckanext/geodatagov/tests/test_db_solr_sync.py

@@ -1,16 +1,13 @@
 import json
 import logging
-import os
 
 import pytest
 from ckan.common import config
 from ckan.lib.search.common import make_connection
 import ckan.model as model
 import ckan.lib.search as search
 from ckan.tests import factories
-from ckan.tests.helpers import reset_db
 from click.testing import CliRunner
-from ckan.model.meta import Session, metadata
 
 from ckanext.harvest.model import HarvestObject
 from ckanext.harvest.tests import factories as harvest_factories
@@ -23,16 +20,6 @@
 
 @pytest.mark.usefixtures("with_plugins")
 class TestSolrDBSync(object):
-    @classmethod
-    def setup_class(cls):
-        # https://github.com/ckan/ckan/issues/4764
-        # drop extension postgis so we can reset db
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'drop extension IF EXISTS postgis cascade;'")
-        reset_db()
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'create extension postgis;'")
-        metadata.create_all(bind=Session.bind)
-
-        search.clear_all()
 
     def create_datasets(self):
 

ckanext/geodatagov/tests/test_fix_packages.py

@@ -1,10 +1,7 @@
 import pytest
 import logging
-import os
 
-from ckan.tests.helpers import reset_db
 from ckan.tests import factories
-from ckan.model.meta import Session, metadata
 
 
 log = logging.getLogger(__name__)
@@ -14,12 +11,6 @@ class TestFixPkg(object):
 
     @classmethod
     def setup_class(cls):
-        # https://github.com/ckan/ckan/issues/4764
-        # drop extension postgis so we can reset db
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'drop extension IF EXISTS postgis cascade;'")
-        reset_db()
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'create extension postgis;'")
-        metadata.create_all(bind=Session.bind)
         cls.organization = factories.Organization()
 
     def test_fix_tags(self):

ckanext/geodatagov/tests/test_fix_spatial.py

@@ -11,9 +11,6 @@ class TestSpatialField(object):
 
     @classmethod
     def setup_class(cls):
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'drop extension IF EXISTS postgis cascade;'")
-        helpers.reset_db()
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'create extension postgis;'")
         # echo "Downloading locations table"
         os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'DROP TABLE IF EXISTS locations;'")
         os.system("wget https://github.com/GSA/datagov-deploy/raw/71936f004be1882a506362670b82c710c64ef796/"
@@ -23,8 +20,6 @@ def setup_class(cls):
         # echo "Cleaning"
         os.system("rm -f /tmp/locations.sql.gz")
 
-        metadata.create_all(bind=Session.bind)
-
         cls.user = factories.Sysadmin(name='spatial_user')
 
     def test_numeric_spatial_transformation(self):

ckanext/geodatagov/tests/test_logic.py

@@ -1,37 +1,16 @@
 import json
-import os
 
+from utils import populate_locations_table
 from ckan.tests.helpers import FunctionalTestBase
-import ckan.lib.search as search
 from ckan.tests import factories
-from ckan.tests.helpers import reset_db
-from ckan.model.meta import Session, metadata
 
 from ckanext.geodatagov.logic import rollup_save_action
 
 
 class TestLogic(FunctionalTestBase):
 
-    @classmethod
-    def setup(cls):
-        search.clear_all()
-
     def setup_method(self):
-        # https://github.com/ckan/ckan/issues/4764
-        # drop extension postgis so we can reset db
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'drop extension IF EXISTS postgis cascade;'")
-        reset_db()
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'create extension postgis;'")
-        # echo "Downloading locations table"
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'DROP TABLE IF EXISTS locations;'")
-        os.system("wget https://github.com/GSA/datagov-deploy/raw/71936f004be1882a506362670b82c710c64ef796/"
-                  "ansible/roles/software/ec2/ansible/files/locations.sql.gz -O /tmp/locations.sql.gz")
-        # echo "Creating locations table"
-        os.system("gunzip -c /tmp/locations.sql.gz | PGPASSWORD=ckan psql -h db -U ckan -d ckan -v ON_ERROR_STOP=1")
-        # echo "Cleaning"
-        os.system("rm -f /tmp/locations.sql.gz")
-        # os.system("ckan -c test.ini db upgrade -p harvest")
-        metadata.create_all(bind=Session.bind)
+        populate_locations_table()
 
     def create_datasets(self):
         self.group1 = factories.Group()

ckanext/geodatagov/tests/test_update_geo.py

@@ -1,7 +1,6 @@
 import logging
 
-from ckan.tests.helpers import reset_db
-
+from utils import populate_locations_table
 from ckanext.geodatagov.logic import translate_spatial
 
 
@@ -10,9 +9,8 @@
 
 class TestUpdateGeo(object):
 
-    @classmethod
-    def setup(cls):
-        reset_db()
+    def setup_method(self):
+        populate_locations_table()
 
     def test_translations(self):
         """ test translate_spatial function """

ckanext/geodatagov/tests/test_waf-collection.py

@@ -1,38 +1,23 @@
 import json
 import logging
 import pytest
-import os
 
-
-from ckan.model.meta import Session, metadata
 from factories import HarvestJobObj, WafCollectionHarvestSourceObj
-import mock_static_file_server
 from ckan import model
 from ckanext.geodatagov.harvesters.waf_collection import WAFCollectionHarvester
 from ckanext.spatial.validation import all_validators
 import ckanext.harvest.model as harvest_model
-
-from ckan.tests.helpers import reset_db, call_action
 from ckan.tests.factories import Organization
 
+from utils import PORT
+
 log = logging.getLogger(__name__)
 
 @pytest.mark.usefixtures("with_plugins")
 class TestWafCollectionHarvester(object):
 
-    @classmethod
-    def setup_class(cls):
-        log.info('Starting mock http server')
-        mock_static_file_server.serve()        
 
     def setup_method(self):
-        # https://github.com/ckan/ckan/issues/4764
-        # drop extension postgis so we can reset db
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'drop extension IF EXISTS postgis cascade;'")
-        reset_db()
-        os.system("PGPASSWORD=ckan psql -h db -U ckan -d ckan -c 'create extension postgis;'")
-        # os.system("ckan -c test.ini db upgrade -p harvest")
-        metadata.create_all(bind=Session.bind)
 
         self.organization = Organization()
 
@@ -104,10 +89,9 @@ def run_import(self):
 
     def get_datasets_from_waf_collection1_sample(self):
         """ harvest waf-collection1/ folder as waf-collection source """
-        url = 'http://127.0.0.1:%s/waf-collection1/index.html' % mock_static_file_server.PORT
+        url = f'http://127.0.0.1:{PORT}/waf-collection1/index.html'
 
-        collection_metadata = "http://127.0.0.1:%s/waf-collection1/cfg/SeriesCollection_tl_2013_county.shp.iso.xml" %\
-            mock_static_file_server.PORT
+        collection_metadata = f"http://127.0.0.1:{PORT}/waf-collection1/cfg/SeriesCollection_tl_2013_county.shp.iso.xml"
         config = '{"collection_metadata_url": "%s", "validator_profiles": ["iso19139ngdc"], "private_datasets": false}' %\
             collection_metadata
         self.run_gather(url=url, source_config=config)