tests: Add fixture to create content directory

The `create_contentdir` function takes the fake channel DB JSON files
and creates a content directory with the channel databases and content
files for testing. That's made available as the `contentdir` pytest
fixture.

A standalone script is provided as a convenience and for exercising the
functionality outside of the test suite.

Author: dbnicholson

kolibri_explore_plugin/test/conftest.py

@@ -5,8 +5,24 @@
 import pytest
 from django.core.management import call_command
 
+from .utils import create_contentdir
+
 
 @pytest.fixture
 def content_data(db):
     """Load test content database fixture"""
     call_command("loaddata", "test-content.json")
+
+
+@pytest.fixture(scope="session")
+def serverdir(tmp_path_factory):
+    """Session scoped server root directory"""
+    return tmp_path_factory.mktemp("server")
+
+
+@pytest.fixture(scope="session")
+def contentdir(serverdir):
+    """Session scoped server content directory"""
+    contentdir = serverdir / "content"
+    create_contentdir(contentdir)
+    return contentdir

kolibri_explore_plugin/test/utils.py

@@ -0,0 +1,103 @@
+# Test utilities
+#
+# Copyright 2023 Endless OS Foundation LLC
+# SPDX-License-Identifier: GPL-2.0-or-later
+import json
+import logging
+import os
+from base64 import b64decode
+from glob import iglob
+from hashlib import md5
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+TESTDIR = Path(__file__).parent.resolve()
+CHANNELSDIR = TESTDIR / "channels"
+
+
+def create_contentdir(content_path, channels_path=CHANNELSDIR):
+    """Create content directory from channel JSON files"""
+    from kolibri.core.content.constants.schema_versions import (
+        CURRENT_SCHEMA_VERSION,
+    )
+    from kolibri.core.content.utils.sqlalchemybridge import Bridge
+
+    databases_path = content_path / "databases"
+    databases_path.mkdir(parents=True, exist_ok=True)
+    storage_path = content_path / "storage"
+    storage_path.mkdir(parents=True, exist_ok=True)
+
+    for json_path in iglob(f"{channels_path}/*.json"):
+        logger.info(f"Loading channel JSON {json_path}")
+        with open(json_path, "r") as f:
+            data = json.load(f)
+
+        channels = data["content_channelmetadata"]
+        if len(channels) != 1:
+            raise ValueError(
+                "Must be one channel in content_channelmetadata table"
+            )
+
+        channel_id = channels[0]["id"]
+        db_path = databases_path / f"{channel_id}.sqlite3"
+        if db_path.exists():
+            logger.info(f"Removing existing channel database {db_path}")
+            db_path.unlink()
+
+        logger.info(f"Creating channel database {db_path}")
+        bridge = Bridge(db_path, schema_version=CURRENT_SCHEMA_VERSION)
+        bridge.Base.metadata.bind = bridge.engine
+        bridge.Base.metadata.create_all()
+
+        # Create the content files from the localfile _content entries.
+        logger.info(f"Creating channel {channel_id} content files")
+        for localfile in data["content_localfile"]:
+            id = localfile["id"]
+            size = localfile["file_size"]
+            ext = localfile["extension"]
+            content = b64decode(localfile.pop("_content"))
+            content_size = len(content)
+            if content_size != size:
+                raise ValueError(
+                    f"Localfile {id} size {size} does not match "
+                    f"content size {content_size}"
+                )
+            content_md5 = md5(content).hexdigest()
+            if content_md5 != id:
+                raise ValueError(
+                    f"Localfile {id} does not match content md5sum "
+                    f"{content_md5}"
+                )
+
+            # If the file already exists, validate its contents. Otherwise,
+            # create it.
+            localfile_dir = storage_path / f"{id[0]}/{id[1]}"
+            localfile_dir.mkdir(parents=True, exist_ok=True)
+            localfile_path = localfile_dir / f"{id}.{ext}"
+            if localfile_path.exists():
+                logger.info(f"Validating content file {localfile_path}")
+                localfile_size = os.path.getsize(localfile_path)
+                if localfile_size != size:
+                    raise ValueError(
+                        f"Localfile {id} size {size} does not match "
+                        f"{localfile_path} size {localfile_size}"
+                    )
+                with open(localfile_path, "rb") as f:
+                    localfile_content = f.read()
+                if localfile_content != content:
+                    raise ValueError(
+                        f"Localfile {id} content does not match "
+                        f"{localfile_path} content"
+                    )
+            else:
+                logger.debug(f"Creating content file {localfile_path}")
+                with open(localfile_path, "wb") as f:
+                    f.write(content)
+
+        # Now fill all the database tables.
+        for table in bridge.Base.metadata.sorted_tables:
+            if data.get(table.name):
+                bridge.connection.execute(table.insert(), data[table.name])
+
+        bridge.end()

scripts/create_contentdir.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+# Copyright 2023 Endless OS Foundation LLC
+# SPDX-License-Identifier: GPL-2.0-or-later
+import os
+from argparse import ArgumentParser
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+from kolibri_explore_plugin.test.utils import create_contentdir
+
+SRCDIR = Path(__file__).parent.parent
+CHANNELSDIR = SRCDIR / "kolibri_explore_plugin/test/channels"
+
+
+def main():
+    ap = ArgumentParser(description="Create channel database from JSON")
+    ap.add_argument(
+        "contentdir",
+        metavar="CONTENTDIR",
+        type=Path,
+        help="content output directory",
+    )
+    ap.add_argument(
+        "-c",
+        "--channelsdir",
+        metavar="CHANNELSDIR",
+        type=Path,
+        default=CHANNELSDIR,
+        help="test channels directory (default: %(default)s)",
+    )
+    args = ap.parse_args()
+
+    # Unfortunately, sqlalchemybridge can't be used without an initialized
+    # Kolibri homedir, so make a temporary one.
+    with TemporaryDirectory(prefix="kolibri-home-") as kolibri_home:
+        os.environ["KOLIBRI_HOME"] = kolibri_home
+        from kolibri.main import initialize
+
+        initialize()
+        create_contentdir(args.contentdir, args.channelsdir)
+
+
+if __name__ == "__main__":
+    main()