Mercurial > hg > orthanc-tests

changeset 909:36d412e3a9f5

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
new tests for LocalAet
author Alain Mazy <am@orthanc.team>
date Mon, 27 Apr 2026 17:10:30 +0200
parents 1c0345bc3721
children f2bde044116b
files GenerateConfigurationForTests.py NewTests/CGet/docker-compose-c-get.yml NewTests/CGet/test_cget.py NewTests/requirements.txt Tests/Run.py Tests/Tests.py Tests/Toolbox.py
diffstat 7 files changed, 119 insertions(+), 18 deletions(-) [+]
line wrap: on
line diff
--- a/GenerateConfigurationForTests.py	Wed Apr 22 19:24:13 2026 +0200
+++ b/GenerateConfigurationForTests.py	Mon Apr 27 17:10:30 2026 +0200
@@ -102,10 +102,10 @@
 
 with open(args.target, 'r') as f:
     # Remove the C++-style comments
-    nocomment = re.sub('//.*$', '', f.read(), 0, re.MULTILINE)
+    nocomment = re.sub(r'//.*$', '', f.read(), 0, re.MULTILINE)
 
     # Remove the C-style comments
-    nocomment = re.sub('/\*.*?\*/', '', nocomment, 0, re.DOTALL | re.MULTILINE)
+    nocomment = re.sub(r'/\*.*?\*/', '', nocomment, 0, re.DOTALL | re.MULTILINE)
 
     config = json.loads(nocomment)
 
@@ -115,7 +115,19 @@
 config['DicomAssociationCloseDelay'] = 0
 config['DicomModalities'] = {
      'orthanctest' : [ 'ORTHANCTEST', ip, 5001 ],
-     'self' : [ 'ORTHANC', '127.0.0.1', 4242 ]
+     'self' : [ 'ORTHANC', '127.0.0.1', 4242 ],
+     'self-with-local-aet': {
+         'AET': 'ORTHANC',
+         'Host': '127.0.0.1',
+         'Port': 4242,
+         'LocalAet': 'SELF-FROM-CONFIG'
+     },
+     'orthanctest-with-local-aet': {
+         'AET': 'ORTHANCTEST',
+         'Host': ip,
+         'Port': 5001,
+         'LocalAet': 'OT-FROM-CONFIG'
+     }
 }
 config['DicomPort'] = args.dicom
 config['HttpCompressionEnabled'] = False
--- a/NewTests/CGet/docker-compose-c-get.yml	Wed Apr 22 19:24:13 2026 +0200
+++ b/NewTests/CGet/docker-compose-c-get.yml	Mon Apr 27 17:10:30 2026 +0200
@@ -33,6 +33,12 @@
               "Port": 4242,
               "Host": "orthanc-b",
               "RetrieveMethod": "C-GET"
+            },
+            "b-with-local-aet-not-declared": {
+              "AET": "ORTHANCB",
+              "Port": 4242,
+              "Host": "orthanc-b",
+              "LocalAet": "FROM-CONFIG"
             }
           }
         }
--- a/NewTests/CGet/test_cget.py	Wed Apr 22 19:24:13 2026 +0200
+++ b/NewTests/CGet/test_cget.py	Mon Apr 27 17:10:30 2026 +0200
@@ -77,5 +77,19 @@
         if oa.is_orthanc_version_at_least(1, 12, 10):
             with self.assertRaises(HttpError) as ex:
                 oa.modalities.get_study(from_modality='b', dicom_id='5.6.7')
-            self.assertEqual(0xc000, ex.exception.dimse_error_status)
+                self.assertEqual(0xc000, ex.exception.error_payload['Content'][0]['DimseErrorStatus'])
             self.assertEqual(0, len(oa.instances.get_all_ids()))       
+
+
+    def test_cget_local_aet(self):
+
+        oa, ob = self.clean_start()
+
+        instances_ids = ob.upload_folder( here / "../../Database/Brainix")
+
+        if oa.is_orthanc_version_at_least(1, 12, 12):
+            # this one fails because the AET is not declared on b
+            with self.assertRaises(HttpError) as ex:
+                oa.modalities.get_study(from_modality='b-with-local-aet-not-declared', dicom_id='2.16.840.1.113669.632.20.1211.10000357775')
+                self.assertEqual(0xc000, ex.exception.error_payload['Content'][0]['DimseErrorStatus'])
+            self.assertEqual(0, len(oa.instances.get_all_ids()))
--- a/NewTests/requirements.txt	Wed Apr 22 19:24:13 2026 +0200
+++ b/NewTests/requirements.txt	Mon Apr 27 17:10:30 2026 +0200
@@ -1,4 +1,4 @@
-orthanc-api-client>=0.22.1
+orthanc-api-client>=0.23.0
 orthanc-tools>=0.16.5
 uvicorn
 fastapi
\ No newline at end of file
--- a/Tests/Run.py	Wed Apr 22 19:24:13 2026 +0200
+++ b/Tests/Run.py	Mon Apr 27 17:10:30 2026 +0200
@@ -108,8 +108,8 @@
 config = re.sub(r'("ExecuteLuaEnabled"\s*:)\s*false', r'\1 true', config)
 config = re.sub(r'("HttpCompressionEnabled"\s*:)\s*true', r'\1 false', config)
 config = re.sub(r'("DicomAssociationCloseDelay"\s*:)\s*[0-9]*', r'\1 0', config)
-config = re.sub(r'("DicomModalities"\s*:)\s*{', r'\1 { "orthanc" : [ "%s", "%s", %s ]' % 
-                (args.aet, args.server, args.dicom), config)
+config = re.sub(r'("DicomModalities"\s*:)\s*{', r'\1 { "orthanc" : [ "%s", "%s", %s ], "orthanc2" : [ "OT-FROM-CONFIG", "%s", %s]' % 
+                (args.aet, args.server, args.dicom, args.server, args.dicom), config)
 
 # New to test transcoding over DICOM (1.7.0)
 config = re.sub(r'("RleTransferSyntaxAccepted"\s*:)\s*true', r'\1 false', config)
--- a/Tests/Tests.py	Wed Apr 22 19:24:13 2026 +0200
+++ b/Tests/Tests.py	Mon Apr 27 17:10:30 2026 +0200
@@ -3262,23 +3262,51 @@
         b = UploadInstance(_REMOTE, 'Knee/T1/IM-0001-0002.dcm')
         c = UploadInstance(_REMOTE, 'Knee/T1/IM-0001-0003.dcm')
 
-        j = DoPost(_REMOTE, '/modalities/orthanctest/store', {
-            'LocalAet' : 'YOP',
-            'Resources' : [ a['ID'], b['ID'] ]
-        })
-
-        self.assertEqual(2, len(DoGet(_LOCAL, '/instances')))
-        self.assertEqual('YOP', DoGet(_LOCAL, '/instances/%s/metadata/RemoteAET' % a['ID']))
-
+        # use the DicomAet from the config file
         DropOrthanc(_LOCAL)
         self.assertEqual(0, len(DoGet(_LOCAL, '/instances')))
 
         j = DoPost(_REMOTE, '/modalities/orthanctest/store', {
-            'Resources' : [ c['ID'] ]
+            'Resources' : [ c['ID'] ],
+            'Synchronous': False
         })
 
+        WaitJobDone(_REMOTE, j['ID'])
         self.assertEqual(1, len(DoGet(_LOCAL, '/instances')))
         self.assertEqual('ORTHANC', DoGet(_LOCAL, '/instances/%s/metadata/RemoteAET' % c['ID']))
+        if IsOrthancVersionAbove(_REMOTE, 1, 12, 12):
+            self.assertEqual('ORTHANC', DoGet(_REMOTE, '/jobs/%s' % j['ID'])['Content']['LocalAet'])
+
+        # use the LocalAet from the "DicomModalities" config
+        if IsOrthancVersionAbove(_REMOTE, 1, 12, 12):
+            DropOrthanc(_LOCAL)
+            self.assertEqual(0, len(DoGet(_LOCAL, '/instances')))
+
+            j = DoPost(_REMOTE, '/modalities/orthanctest-with-local-aet/store', {
+                'Resources' : [ c['ID'] ],
+                'Synchronous': False
+            })
+
+            WaitJobDone(_REMOTE, j['ID'])
+            self.assertEqual(1, len(DoGet(_LOCAL, '/instances')))
+            self.assertEqual('OT-FROM-CONFIG', DoGet(_LOCAL, '/instances/%s/metadata/RemoteAET' % c['ID']))
+            self.assertEqual('OT-FROM-CONFIG', DoGet(_REMOTE, '/jobs/%s' % j['ID'])['Content']['LocalAet'])
+
+
+        # use the LocalAet from the payload
+        DropOrthanc(_LOCAL)
+        j = DoPost(_REMOTE, '/modalities/orthanctest-with-local-aet/store', {
+            'LocalAet' : 'FROM-PAYLOAD',
+            'Resources' : [ a['ID'], b['ID'] ],
+            'Synchronous': False
+        })
+
+        WaitJobDone(_REMOTE, j['ID'])
+        self.assertEqual(2, len(DoGet(_LOCAL, '/instances')))
+        if IsOrthancVersionAbove(_REMOTE, 1, 12, 12):
+            self.assertEqual('FROM-PAYLOAD', DoGet(_LOCAL, '/instances/%s/metadata/RemoteAET' % a['ID']))
+            self.assertEqual('FROM-PAYLOAD', DoGet(_REMOTE, '/jobs/%s' % j['ID'])['Content']['LocalAet'])
+
 
         DropOrthanc(_REMOTE)        
         DropOrthanc(_LOCAL)        
@@ -4233,6 +4261,43 @@
             self.assertEqual(4, len(DoGet(_REMOTE, '/instances')))
 
 
+    def test_move_scu_local_aet(self):
+        # test localAet wrt Move
+        if IsOrthancVersionAbove(_REMOTE, 1, 12, 12):
+            DropOrthanc(_LOCAL)
+            DropOrthanc(_REMOTE)
+            DropOrthancJobs(_REMOTE)
+            UploadInstance(_REMOTE, 'Knee/T2/IM-0001-0001.dcm')
+
+            with open(os.devnull, 'w') as FNULL:
+                try:
+                    subprocess.check_call([ 
+                            FindExecutable('movescu'), 
+                            '--move', 'ORTHANCTEST2',         # Target AET (use this one to force usage of LocalAet)
+                            '--call', _REMOTE['DicomAet'],     # Called AET (i.e. Orthanc)
+                            '--aetitle', _LOCAL['DicomAet'],   # Calling AET (i.e. storescp)
+                            _REMOTE['Server'], 
+                            str(_REMOTE['DicomPort']),
+                            '--study', '-k', 'QueryRetrieveLevel=Study',
+                            '-k', 'StudyInstanceUID=2.16.840.1.113669.632.20.121711.10000160881'
+                        ],
+                        stderr = FNULL)
+
+                except subprocess.CalledProcessError as e:
+                    print('movescu failed with error code: %s' % str(e.returncode))
+                    raise e
+
+            jobCompleted = False
+            while not jobCompleted:
+                allJobs = DoGet(_REMOTE, '/jobs?expand')
+                jobCompleted = len(allJobs) == 1 and allJobs[0]['State'] == 'Success'
+                time.sleep(0.1)
+
+            allInstances = DoGet(_LOCAL, '/instances')
+            self.assertEqual(1, len(allInstances))
+            self.assertEqual('OT-FROM-CONFIG', DoGet(_LOCAL, '/instances/%s/metadata/RemoteAET' % allInstances[0]))
+            self.assertEqual('OT-FROM-CONFIG', DoGet(_REMOTE, '/jobs/%s' % allJobs[0]['ID'])['Content']['LocalAet'])
+
 
     def test_reconstruct_json(self):
         self.assertEqual(0, len(DoGet(_REMOTE, '/patients')))
--- a/Tests/Toolbox.py	Wed Apr 22 19:24:13 2026 +0200
+++ b/Tests/Toolbox.py	Mon Apr 27 17:10:30 2026 +0200
@@ -205,6 +205,10 @@
     for s in DoGet(orthanc, '/patients'):
         DoDelete(orthanc, '/patients/%s' % s)
 
+def DropOrthancJobs(orthanc):
+    for s in DoGet(orthanc, '/jobs'):
+        DoDelete(orthanc, '/jobs/%s' % s)
+
 def InstallLuaScriptFromPath(orthanc, path):
     with open(GetDatabasePath(path), 'r') as f:
         InstallLuaScript(orthanc, f.read())
@@ -257,9 +261,9 @@
             return
         time.sleep(0.1)
 
-def WaitJobDone(orthanc, job):
+def WaitJobDone(orthanc, jobId):
     while True:
-        s = DoGet(orthanc, '/jobs/%s' % job) ['State']
+        s = DoGet(orthanc, '/jobs/%s' % jobId) ['State']
 
         if s == 'Success':
             return True