Add tests

.github/workflows/tests.yml

@@ -0,0 +1,29 @@
+name: Run tests
+on:
+  push
+
+jobs:
+  tests:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-python@v4
+        with:
+          python-version: "3.11"
+      - uses: actions/setup-node@v3
+        with:
+          node-version: "18.x"
+      - name: Install pip and poetry
+        run: pip install --upgrade pip poetry
+      - uses: actions/cache@v3
+        with:
+          path: ~/.cache/pypoetry/virtualenvs
+          key: ${{ runner.os }}-poetry-${{ hashFiles('src/poetry.lock') }}
+      - name: Install Poetry dependencies
+        run: cd src && poetry install
+      - name: Install Node dependencies
+        run: cd src/frontend && npm install
+      - name: Build frontend
+        run: cd src/frontend && npm run build
+      - name: Run tests
+        run: cd src && poetry run pytest -v

src/app.py

@@ -6,11 +6,14 @@
 from pathlib import Path
 from flask import Flask, jsonify, render_template, abort, send_file, request
 
-from common import default_structures_path, builtin_structures_path
+from common import (
+    get_blueleaks_path,
+    get_dbs_path,
+    get_structures_path,
+    get_default_structures_path,
+    get_builtin_structures_path,
+)
 
-blueleaks_path = os.environ.get("BLE_BLUELEAKS_PATH")
-dbs_path = os.environ.get("BLE_DATABASES_PATH")
-structures_path = os.environ.get("BLE_STRUCTURES_PATH")
 
 app = Flask(__name__, static_folder="frontend/dist/assets", static_url_path="/assets")
 
@@ -23,7 +26,7 @@
 
 def get_structure(site):
     if site not in structures:
-        with open(os.path.join(structures_path, f"{site}.json")) as f:
+        with open(os.path.join(get_structures_path(), f"{site}.json")) as f:
             structures[site] = json.load(f)
 
     return structures[site]
@@ -40,12 +43,12 @@ def humansize(nbytes):
 
 
 def get_database_filename(site):
-    return os.path.join(dbs_path, f"{site}.sqlite3")
+    return os.path.join(get_dbs_path(), f"{site}.sqlite3")
 
 
 def get_all_sites():
     all_sites = []
-    for filename in os.listdir(default_structures_path):
+    for filename in os.listdir(get_default_structures_path()):
         if filename.endswith(".json"):
             all_sites.append(filename[:-5])
     all_sites.sort()
@@ -54,7 +57,7 @@ def get_all_sites():
 
 def get_implemented_sites():
     implemented_sites = []
-    for filename in os.listdir(structures_path):
+    for filename in os.listdir(get_structures_path()):
         if filename.endswith(".json"):
             implemented_sites.append(filename[:-5])
     implemented_sites.sort()
@@ -227,23 +230,23 @@ def render_frontend():
 @app.route("/<path:path>")
 def catch_all(path):
     # Make sure BlueLeaks data is all there
-    if not os.path.isdir(blueleaks_path):
+    if not os.path.isdir(get_blueleaks_path()):
         return render_template(
             "error.html",
-            error_message=f"The BlueLeaks data folder {blueleaks_path} isn't a folder",
+            error_message=f"The BlueLeaks data folder {get_blueleaks_path()} isn't a folder",
         )
     missing_dbs = []
     missing_sites = []
-    sites = os.listdir(blueleaks_path)
+    sites = os.listdir(get_blueleaks_path())
     for site in get_all_sites():
         if site not in sites:
             missing_sites.append(site)
-        if not os.path.isfile(os.path.join(dbs_path, f"{site}.sqlite3")):
+        if not os.path.isfile(os.path.join(get_dbs_path(), f"{site}.sqlite3")):
             missing_dbs.append(site)
     if len(missing_sites) > 0:
         return render_template(
             "error.html",
-            error_message=f"Can't find the unzipped BlueLeaks dataset. Fix the volume that maps to {blueleaks_path} in your docker-compose.yaml.",
+            error_message=f"Can't find the unzipped BlueLeaks dataset. Fix the volume that maps to {get_blueleaks_path()} in your docker-compose.yaml.",
         )
     if len(missing_dbs) > 0:
         return render_template(
@@ -256,7 +259,7 @@ def catch_all(path):
         listing_path = path[len("blueleaks-data") :]
         if listing_path == "":
             listing_path = "/"
-        filename = os.path.join(blueleaks_path, listing_path.lstrip("/"))
+        filename = os.path.join(get_blueleaks_path(), listing_path.lstrip("/"))
         if os.path.exists(filename):
             if os.path.isdir(filename):
                 # Render directory listing
@@ -270,7 +273,7 @@ def catch_all(path):
                             "link": os.path.join(
                                 "/blueleaks-data",
                                 os.path.dirname(listing_path).lstrip("/"),
-                            ),
+                            ).replace("\\", "/"),
                         }
                     )
 
@@ -281,7 +284,7 @@ def catch_all(path):
                                 "name": name,
                                 "link": os.path.join(
                                     "/blueleaks-data", listing_path.lstrip("/"), name
-                                ),
+                                ).replace("\\", "/"),
                             }
                         )
                     else:
@@ -291,7 +294,7 @@ def catch_all(path):
                                 "name": name,
                                 "link": os.path.join(
                                     "/blueleaks-data", listing_path.lstrip("/"), name
-                                ),
+                                ).replace("\\", "/"),
                                 "size": humansize(size_bytes),
                             }
                         )
@@ -340,15 +343,15 @@ def api_structure_create(site):
         return jsonify({"error": True, "error_message": "Invalid site"})
 
     # Is this site already implemented?
-    if os.path.exists(os.path.join(structures_path, f"{site}.json")):
+    if os.path.exists(os.path.join(get_structures_path(), f"{site}.json")):
         return jsonify(
             {"error": True, "error_message": "That site is already implemented"}
         )
 
     # Copy the default structure
     shutil.copyfile(
-        os.path.join(default_structures_path, f"{site}.json"),
-        os.path.join(structures_path, f"{site}.json"),
+        os.path.join(get_default_structures_path(), f"{site}.json"),
+        os.path.join(get_structures_path(), f"{site}.json"),
     )
     return jsonify({"error": False})
 
@@ -363,21 +366,21 @@ def api_structure(site):
         return jsonify({"error": True, "error_message": "Invalid site"})
 
     # Has it been implemented?
-    if not os.path.exists(os.path.join(structures_path, f"{site}.json")):
+    if not os.path.exists(os.path.join(get_structures_path(), f"{site}.json")):
         return jsonify(
             {"error": True, "error_message": "That site hasn't been implemented"}
         )
 
     if request.method == "GET":
         # Return the structure
-        with open(os.path.join(structures_path, f"{site}.json")) as f:
+        with open(os.path.join(get_structures_path(), f"{site}.json")) as f:
             structure = json.load(f)
         return jsonify({"error": False, "structure": structure})
     elif request.method == "POST":
         # Save the structure
         structure = request.json
         structures[site] = structure
-        with open(os.path.join(structures_path, f"{site}.json"), "w") as f:
+        with open(os.path.join(get_structures_path(), f"{site}.json"), "w") as f:
             f.write(json.dumps(structure, indent=4))
         return jsonify({"error": False})
     else:
@@ -581,10 +584,10 @@ def api_join(site, table, join_name, item_id):
 # Copy the builtin structures, if there isn't already a structure there
 for site in get_all_sites():
     if os.path.exists(
-        os.path.join(builtin_structures_path, f"{site}.json")
-    ) and not os.path.exists(os.path.join(structures_path, f"{site}.json")):
+        os.path.join(get_builtin_structures_path(), f"{site}.json")
+    ) and not os.path.exists(os.path.join(get_structures_path(), f"{site}.json")):
         print(f"copying built-in structure: {site}")
         shutil.copyfile(
-            os.path.join(builtin_structures_path, f"{site}.json"),
-            os.path.join(structures_path, f"{site}.json"),
+            os.path.join(get_builtin_structures_path(), f"{site}.json"),
+            os.path.join(get_structures_path(), f"{site}.json"),
         )

src/build-structures.py

@@ -3,7 +3,7 @@
 import csv
 import click
 
-from common import sanitize_field_name, default_structures_path
+from common import sanitize_field_name, get_default_structures_path
 
 blueleaks_path = os.environ.get("BLE_BLUELEAKS_PATH")
 
@@ -61,7 +61,7 @@ def build_structure():
                     "joins": [],
                 }
 
-            json_filename = os.path.join(default_structures_path, f"{site}.json")
+            json_filename = os.path.join(get_default_structures_path(), f"{site}.json")
             with open(json_filename, "w") as f:
                 f.write(json.dumps(structure, indent=4))
             click.secho(f"Wrote {json_filename}", dim=True)

src/common.py

@@ -1,9 +1,44 @@
 import os
 
-builtin_structures_path = "/var/blueleaks-explorer/structures-builtin"
 
-default_structures_path = "/var/blueleaks-explorer/structures-default"
-os.makedirs(default_structures_path, exist_ok=True)
+def get_blueleaks_path():
+    return os.environ.get("BLE_BLUELEAKS_PATH", "")
+
+
+def get_dbs_path():
+    return os.environ.get("BLE_DATABASES_PATH", "")
+
+
+def get_structures_path():
+    return os.environ.get("BLE_STRUCTURES_PATH", "")
+
+
+def get_builtin_structures_path():
+    path = os.environ.get(
+        "BLE_STRUCTURES_BUILTIN_PATH", "/var/blueleaks-explorer/structures-builtin"
+    )
+
+    try:
+        os.makedirs(path, exist_ok=True)
+    except:
+        path = "/tmp/blueleaks-explorer/structures-builtin"
+        os.makedirs(path, exist_ok=True)
+
+    return path
+
+
+def get_default_structures_path():
+    path = os.environ.get(
+        "BLE_STRUCTURES_DEFAULT_PATH", "/var/blueleaks-explorer/structures-default"
+    )
+
+    try:
+        os.makedirs(path, exist_ok=True)
+    except:
+        path = "/tmp/blueleaks-explorer/structures-default"
+        os.makedirs(path, exist_ok=True)
+
+    return path
 
 
 def sanitize_field_name(table):

src/frontend/index.html

@@ -14,7 +14,7 @@
     <div id="app"></div>
     <script type="module" src="/src/main.js"></script>
     <noscript>
-        <p><strong>The BlueLeaks Explorer requires javascript.</strong></p>
+        <p><strong>BlueLeaks Explorer requires javascript.</strong></p>
         <p>If you're in Tor Browser, click the shield in the top-right and change the security level from Safest to
             Safer or Standard.</p>
     </noscript>

src/initialize.py

@@ -4,10 +4,12 @@
 import sqlite3
 import click
 
-from common import sanitize_field_name, default_structures_path
-
-blueleaks_path = os.environ.get("BLE_BLUELEAKS_PATH")
-dbs_path = os.environ.get("BLE_DATABASES_PATH")
+from common import (
+    sanitize_field_name,
+    get_default_structures_path,
+    get_blueleaks_path,
+    get_dbs_path,
+)
 
 
 def exec_sql(c, sql):
@@ -40,7 +42,7 @@ def load_file(path):
         site = structure["name"]
 
         # Start the database
-        database_filename = os.path.join(dbs_path, f"{site}.sqlite3")
+        database_filename = os.path.join(get_dbs_path(), f"{site}.sqlite3")
         if os.path.exists(database_filename):
             click.secho(f"{database_filename} already exists so skipping", dim=True)
             return
@@ -51,7 +53,7 @@ def load_file(path):
         for table in structure["tables"]:
             progress(site, table)
 
-            csv_filename = os.path.join(blueleaks_path, site, f"{table}.csv")
+            csv_filename = os.path.join(get_blueleaks_path(), site, f"{table}.csv")
             with open(csv_filename) as csv_file:
                 reader = csv.DictReader(csv_file)
 
@@ -104,10 +106,10 @@ def is_struct_json(filename):
 
 
 def import_data():
-    for filename in os.listdir(default_structures_path):
+    for filename in os.listdir(get_default_structures_path()):
         if is_struct_json(filename):
             load_file(
-                os.path.join(default_structures_path, filename),
+                os.path.join(get_default_structures_path(), filename),
             )