New smoke tests for catalog operations

tests/catalog/create_test.py

@@ -0,0 +1,8 @@
+from tests import dds, DDS
+from tests.fileutil import ensure_dir
+
+
+def test_create_catalog(dds: DDS):
+    dds.scope.enter_context(ensure_dir(dds.build_dir))
+    dds.catalog_create(dds.build_dir / 'cat.db')
+    assert (dds.build_dir / 'cat.db').is_file()

tests/catalog/import_test.py

@@ -0,0 +1,30 @@
+import json
+
+from tests import dds, DDS
+from tests.fileutil import ensure_dir
+
+
+def test_import_json(dds: DDS):
+    dds.scope.enter_context(ensure_dir(dds.build_dir))
+    cat_path = dds.build_dir / 'catalog.db'
+    dds.catalog_create(cat_path)
+
+    json_fpath = dds.build_dir / 'data.json'
+    import_data = {
+        'version': 1,
+        'packages': {
+            'foo': {
+                '1.2.4': {
+                    'git': {
+                        'url': 'http://example.com',
+                        'ref': 'master',
+                    },
+                    'depends': {},
+                },
+            },
+        },
+    }
+    dds.scope.enter_context(
+        dds.set_contents(json_fpath,
+                         json.dumps(import_data).encode()))
+    dds.catalog_import(cat_path, json_fpath)

tests/dds.py

@@ -51,7 +51,7 @@ class DDS:
     def run(self, cmd: proc.CommandLine, *,
             cwd: Path = None) -> subprocess.CompletedProcess:
         cmdline = list(proc.flatten_cmd(cmd))
-        res = self.run_unchecked(cmd)
+        res = self.run_unchecked(cmd, cwd=cwd)
         if res.returncode != 0:
             raise subprocess.CalledProcessError(
                 res.returncode, [self.dds_exe] + cmdline, res.stdout)
@@ -143,6 +143,27 @@ class DDS:
                 f'We don\'t know the executable suffix for the platform "{os.name}"'
             )
 
+    def catalog_create(self, path: Path) -> subprocess.CompletedProcess:
+        return self.run(['catalog', 'create', f'--catalog={path}'],
+                        cwd=self.test_dir)
+
+    def catalog_import(self, path: Path,
+                       json_path: Path) -> subprocess.CompletedProcess:
+        return self.run([
+            'catalog',
+            'import',
+            f'--catalog={path}',
+            json_path,
+        ])
+
+    def catalog_get(self, path: Path, req: str) -> subprocess.CompletedProcess:
+        return self.run([
+            'catalog',
+            'get',
+            f'--catalog={path}',
+            req,
+        ])
+
     def set_contents(self, path: Union[str, Path],
                      content: bytes) -> ContextManager[Path]:
         return fileutil.set_contents(self.source_root / path, content)