unit test for download module

ckanext/feedback/controllers/download.py

@@ -5,10 +5,6 @@
 
 
 class DownloadController:
-    @staticmethod
-    def increment_resource_downloads(resource_id):
-        increment_resource_downloads(resource_id)
-
     # extend default download function to count when a resource is downloaded
     @staticmethod
     def extended_download(package_type, id, resource_id, filename=None):

ckanext/feedback/tests/config/test.ini

@@ -4,12 +4,14 @@ smtp_server = localhost
 error_email_from = ckan@localhost
 
 [app:main]
-use = config:../ckan/test-core.ini
+use = config:/usr/lib/ckan/venv/src/ckan/test-core.ini
 
 # Insert any custom config settings to be used when running your extension's
 # tests here. These will override the one defined in CKAN core's test-core.ini
 ckan.plugins = feedback
-
+sqlalchemy.url = postgresql://ckan:ckan@db/ckan_test
+datastore.write.url = postgresql://ckan:ckan@db/datastore_test
+datastore.read.url = postgresql://datastore_ro:ckan@db/datastore_test 
 
 # Logging configuration
 [loggers]

ckanext/feedback/tests/controllers/test_download.py

@@ -0,0 +1,59 @@
+from unittest.mock import patch
+
+import pytest
+from ckan import model
+from ckan.tests import factories
+from flask import Flask
+
+from ckanext.feedback.command.feedback import (
+    create_download_tables,
+    create_resource_tables,
+    create_utilization_tables,
+    get_engine,
+)
+from ckanext.feedback.controllers.download import DownloadController
+from ckanext.feedback.models.download import DownloadSummary
+from ckanext.feedback.models.session import session
+
+
+def get_downloads(resource_id):
+    count = (
+        session.query(DownloadSummary.download)
+        .filter(DownloadSummary.resource_id == resource_id)
+        .scalar()
+    )
+    return count
+
+
+@pytest.mark.usefixtures('clean_db', 'with_plugins', 'with_request_context')
+class TestDownloadController:
+    @classmethod
+    def setup_class(cls):
+        model.repo.init_db()
+        engine = get_engine('db', '5432', 'ckan_test', 'ckan', 'ckan')
+        create_utilization_tables(engine)
+        create_resource_tables(engine)
+        create_download_tables(engine)
+
+    def setup_method(self, method):
+        self.app = Flask(__name__)
+
+    @patch('ckanext.feedback.controllers.download.download')
+    def test_extended_download(self, download):
+        resource = factories.Resource()
+        with self.app.test_request_context(headers={'Sec-Fetch-Dest': 'document'}):
+            DownloadController.extended_download(
+                'package_type', resource['package_id'], resource['id'], None
+            )
+            assert get_downloads(resource['id']) == 1
+            assert download
+
+    @patch('ckanext.feedback.controllers.download.download')
+    def test_extended_download_with_preview(self, download):
+        resource = factories.Resource()
+        with self.app.test_request_context(headers={'Sec-Fetch-Dest': 'image'}):
+            DownloadController.extended_download(
+                'package_type', resource['package_id'], resource['id'], None
+            )
+            assert get_downloads(resource['id']) is None
+            assert download

ckanext/feedback/tests/services/download/test_summary.py

@@ -0,0 +1,70 @@
+import pytest
+from ckan import model
+from ckan.tests import factories
+
+from ckanext.feedback.command.feedback import (
+    create_download_tables,
+    create_resource_tables,
+    create_utilization_tables,
+    get_engine,
+)
+from ckanext.feedback.models.download import DownloadSummary
+from ckanext.feedback.models.session import session
+from ckanext.feedback.services.download.summary import (
+    get_package_downloads,
+    get_resource_downloads,
+    increment_resource_downloads,
+)
+
+
+def get_downloads(resource_id):
+    count = (
+        session.query(DownloadSummary.download)
+        .filter(DownloadSummary.resource_id == resource_id)
+        .scalar()
+    )
+    return count
+
+
+@pytest.mark.usefixtures('clean_db', 'with_plugins', 'with_request_context')
+class TestDownloadServices:
+    @classmethod
+    def setup_class(cls):
+        model.repo.init_db()
+        engine = get_engine('db', '5432', 'ckan_test', 'ckan', 'ckan')
+        create_utilization_tables(engine)
+        create_resource_tables(engine)
+        create_download_tables(engine)
+
+    def test_increment_resource_downloads(self):
+        resource = factories.Resource()
+        increment_resource_downloads(resource['id'])
+        assert get_downloads(resource['id']) == 1
+        increment_resource_downloads(resource['id'])
+        assert get_downloads(resource['id']) == 2
+
+    def test_get_package_download(self):
+        resource = factories.Resource()
+        assert get_package_downloads(resource['package_id']) == 0
+        download_summary = DownloadSummary(
+            id=str('test_id'),
+            resource_id=resource['id'],
+            download=1,
+            created='2023-03-31 01:23:45.123456',
+            updated='2023-03-31 01:23:45.123456',
+        )
+        session.add(download_summary)
+        assert get_package_downloads(resource['package_id']) == 1
+
+    def test_get_resource_download(self):
+        resource = factories.Resource()
+        assert get_resource_downloads(resource['id']) == 0
+        download_summary = DownloadSummary(
+            id=str('test_id'),
+            resource_id=resource['id'],
+            download=1,
+            created='2023-03-31 01:23:45.123456',
+            updated='2023-03-31 01:23:45.123456',
+        )
+        session.add(download_summary)
+        assert get_resource_downloads(resource['id']) == 1