feat(medcat-trainer): scaffold for sourcing models via a medcattery instance

medcat-trainer/webapp/api/api/extensions.py

@@ -88,6 +88,13 @@
 #: kwargs: ``user`` (User), ``id_token`` (dict), ``created`` (bool).
 user_oidc_resolved = Signal()
 
+#: Sent after :func:`api.model_import.import_model_pack` successfully registers
+#: a model pack.
+#: kwargs: ``model_pack`` (ModelPack), ``concept_db`` (ConceptDB or None),
+#:         ``user`` (User or None), ``description`` (str or None),
+#:         ``source_uri`` (str or None).
+model_pack_imported = Signal()
+
 
 # ---------------------------------------------------------------------------
 # Signal dispatch (plugin-isolating)
@@ -290,6 +297,7 @@ def clear_registries() -> None:
     "project_group_created",
     "project_group_updated",
     "user_oidc_resolved",
+    "model_pack_imported",
     "dispatch",
     "register_permission_hook",
     "get_permission_hooks",

medcat-trainer/webapp/api/api/tests/test_model_import.py

@@ -0,0 +1,63 @@
+import os
+import tempfile
+from unittest.mock import patch
+
+from django.conf import settings
+from django.contrib.auth.models import User
+from django.test import TestCase, override_settings
+
+from api.extensions import model_pack_imported
+from api.model_import import ImportModelPackError, import_model_pack
+from api.models import ConceptDB, ModelPack
+
+
+@override_settings(MEDIA_ROOT=tempfile.mkdtemp())
+class ImportModelPackTests(TestCase):
+    def setUp(self):
+        self.user = User.objects.create_user(username='importer', password='pass')
+        self.src_zip = os.path.join(settings.MEDIA_ROOT, 'incoming.zip')
+        with open(self.src_zip, 'wb') as fh:
+            fh.write(b'fake-model-pack-zip')
+
+    def test_rejects_empty_name(self):
+        with self.assertRaises(ImportModelPackError):
+            import_model_pack(self.src_zip, name='   ')
+
+    def test_rejects_duplicate_name(self):
+        existing = ModelPack(name='snomed-v1')
+        existing.save(skip_load=True)
+        with self.assertRaises(ImportModelPackError):
+            import_model_pack(self.src_zip, name='snomed-v1')
+
+    @patch('api.model_import.dispatch')
+    def test_import_from_path_registers_model_pack(self, mock_dispatch):
+        from django.db import models as django_models
+
+        def fake_save(self, *args, **kwargs):
+            """Stand in for ModelPack.save() without unpacking a real archive."""
+            cdb = ConceptDB(name=f'{self.name}_CDB', cdb_file='imported/cdb.dat')
+            cdb.save(skip_load=True)
+            self.concept_db = cdb
+            django_models.Model.save(self)
+
+        with patch.object(ModelPack, 'save', autospec=True, side_effect=fake_save):
+            model_pack = import_model_pack(
+                self.src_zip,
+                name='medcattery-snomed',
+                user=self.user,
+                description='Pulled from MedCATtery',
+                source_uri='https://medcattery.example/models/snomed/3',
+            )
+
+        self.assertEqual(model_pack.name, 'medcattery-snomed')
+        self.assertEqual(model_pack.last_modified_by, self.user)
+        self.assertTrue(model_pack.model_pack.name.startswith('modelpacks/medcattery-snomed-'))
+        self.assertTrue(os.path.exists(model_pack.model_pack.path))
+        mock_dispatch.assert_called_once()
+        signal, kwargs = mock_dispatch.call_args.args[0], mock_dispatch.call_args.kwargs
+        self.assertIs(signal, model_pack_imported)
+        self.assertEqual(kwargs['model_pack'], model_pack)
+        self.assertEqual(kwargs['concept_db'], model_pack.concept_db)
+        self.assertEqual(kwargs['user'], self.user)
+        self.assertEqual(kwargs['description'], 'Pulled from MedCATtery')
+        self.assertEqual(kwargs['source_uri'], 'https://medcattery.example/models/snomed/3')