applications: refresh selected elements on scan (#71251)

hobo/applications/models.py

@@ -106,10 +106,33 @@ class Application(models.Model):
             self.slug = slug
         super().save(*args, **kwargs)
 
-    def scandeps(self):
+    def refresh_elements(self):
         self.relation_set.filter(auto_dependency=True).delete()
+        remote_elements = {}
+        for object_type in get_object_types():
+            if object_type.get('minor'):
+                continue
+            url = object_type['urls']['list']
+            response = requests.get(url)
+            if not response.ok:
+                raise ScanError(
+                    _('Failed to get elements of type %s (%s)' % (object_type['id'], response.status_code))
+                )
+            remote_elements[object_type['id']] = {x['id']: x for x in response.json()['data']}
         relations = self.relation_set.select_related('element')
         elements = {(x.element.type, x.element.slug): (x.element, x) for x in relations}
+        for element, relation in elements.values():
+            if remote_elements[element.type].get(element.slug):
+                remote_element = remote_elements[element.type][element.slug]
+                if element.name != remote_element['text'] or element.cache != remote_element:
+                    element.name = remote_element['text']
+                    element.cache = remote_element
+                    element.save()
+                    elements[(element.type, element.slug)] = (element, relation)
+        return elements
+
+    def scandeps(self):
+        elements = self.refresh_elements()
         finished = False
         while not finished:
             finished = True

tests/test_application.py

@@ -189,6 +189,8 @@ def mocked_http(url, request):
     if url.path == '/api/export-import/bundle-import/':
         return {'content': '{}', 'status_code': 200}
 
+    return {'content': json.dumps({'data': []}), 'status_code': 200}
+
 
 @pytest.mark.parametrize('analyze', [True, False])
 def test_create_application(app, admin_user, settings, analyze):
@@ -511,6 +513,135 @@ def test_scandeps_on_unknown_element(app, admin_user, settings):
         assert element.error_status is None
 
 
+def test_scandeps_on_renamed_element(app, admin_user, settings):
+    Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')
+
+    settings.KNOWN_SERVICES = {
+        'wcs': {
+            'foobar': {
+                'title': 'Foobar',
+                'url': 'https://wcs.example.invalid/',
+                'orig': 'example.org',
+                'secret': 'xxx',
+            }
+        }
+    }
+
+    application = Application.objects.create(name='Test', slug='test')
+    element = Element.objects.create(
+        type='forms',
+        slug='test-form',
+        name='Test 1 form',
+        cache={
+            'urls': {'redirect': 'https://wcs.example.invalid/api/export-import/forms/test-form/redirect/'}
+        },
+    )
+    Relation.objects.create(application=application, element=element)
+    element2 = Element.objects.create(
+        type="cards",
+        slug="test-card",
+        name="Test Card",
+        cache={
+            "urls": {
+                "export": "https://wcs.example.invalid/api/export-import/cards/test-card/",
+                "dependencies": "https://wcs.example.invalid/api/export-import/cards/test-card/dependencies/",
+            }
+        },
+    )
+    Relation.objects.create(application=application, element=element2, auto_dependency=True)
+
+    def response_content(url, request):
+        if url.path == '/api/export-import/forms/':
+            return {
+                'content': {
+                    "data": [
+                        {
+                            "id": "test-form",
+                            "text": "Test Form (renamed)",
+                            "type": "forms",
+                            "urls": {
+                                "export": "https://wcs.example.invalid/api/export-import/forms/test-form/",
+                                "dependencies": "https://wcs.example.invalid/api/export-import/forms/test-form/dependencies/",
+                                "redirect": "https://wcs.example.invalid/api/export-import/forms/test-form/redirect/",
+                            },
+                        },
+                    ]
+                },
+                'status_code': 200,
+            }
+        if url.path == '/api/export-import/forms/test-form/dependencies/':
+            return {
+                'content': {
+                    "data": [
+                        {
+                            "id": "test-card",
+                            "text": "Test Card (renamed)",
+                            "type": "cards",
+                            "urls": {
+                                "dependencies": "https://wcs.example.invalid/api/export-import/cards/test-card/dependencies/",
+                            },
+                        }
+                    ]
+                },
+                'status_code': 200,
+            }
+        return mocked_http(url, request)
+
+    login(app)
+    with HTTMock(response_content):
+        app.get('/applications/manifest/test/scandeps/').follow()
+        job = AsyncJob.objects.latest('pk')
+        assert job.status == 'completed'
+        element.refresh_from_db()
+        assert element.name == 'Test Form (renamed)'
+        assert element.cache == {
+            "id": "test-form",
+            "text": "Test Form (renamed)",
+            "type": "forms",
+            "urls": {
+                "export": "https://wcs.example.invalid/api/export-import/forms/test-form/",
+                "dependencies": "https://wcs.example.invalid/api/export-import/forms/test-form/dependencies/",
+                "redirect": "https://wcs.example.invalid/api/export-import/forms/test-form/redirect/",
+            },
+        }
+        element2 = Element.objects.get(relation__auto_dependency=True)
+        assert element2.name == 'Test Card (renamed)'
+        assert element2.cache == {
+            "id": "test-card",
+            "text": "Test Card (renamed)",
+            "type": "cards",
+            "urls": {
+                "dependencies": "https://wcs.example.invalid/api/export-import/cards/test-card/dependencies/",
+            },
+        }
+
+    def response_content(url, request):
+        if url.path == '/api/export-import/forms/':
+            return {'status_code': 404}
+        return mocked_http(url, request)
+
+    with HTTMock(response_content):
+        with pytest.raises(ScanError) as e:
+            app.get('/applications/manifest/test/scandeps/').follow()
+        assert str(e.value) == 'Failed to get elements of type forms (404)'
+        job = AsyncJob.objects.latest('pk')
+        assert job.status == 'failed'
+        assert job.exception == 'Failed to get elements of type forms (404)'
+
+    def response_content(url, request):
+        if url.path == '/api/export-import/forms/':
+            return {'status_code': 500}
+        return mocked_http(url, request)
+
+    with HTTMock(response_content):
+        with pytest.raises(ScanError) as e:
+            app.get('/applications/manifest/test/scandeps/').follow()
+        assert str(e.value) == 'Failed to get elements of type forms (500)'
+        job = AsyncJob.objects.latest('pk')
+        assert job.status == 'failed'
+        assert job.exception == 'Failed to get elements of type forms (500)'
+
+
 @pytest.mark.parametrize('editable', [True, False])
 def test_redirect_application_element(app, admin_user, settings, editable):
     Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')