129 lines
4.8 KiB
Nix
129 lines
4.8 KiB
Nix
# Shared mock service generators for arr-init NixOS tests.
|
|
#
|
|
# Usage (from a test file):
|
|
# let mocks = import ./lib/mocks.nix { inherit pkgs; };
|
|
# in { systemd.services.mock-qbittorrent = mocks.mkMockQbittorrent { }; }
|
|
{ pkgs }:
|
|
{
|
|
# Mock qBittorrent WebUI API.
|
|
#
|
|
# Args:
|
|
# port - TCP port (default 6011)
|
|
# initialCategories - Nix attrset seeded as the CATEGORIES dict
|
|
# before - systemd Before= list
|
|
mkMockQbittorrent =
|
|
{
|
|
port ? 6011,
|
|
initialCategories ? { },
|
|
before ? [ ],
|
|
}:
|
|
let
|
|
categoriesJson = builtins.toJSON initialCategories;
|
|
mockScript = pkgs.writeScript "mock-qbittorrent.py" ''
|
|
import json
|
|
from http.server import HTTPServer, BaseHTTPRequestHandler
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
CATEGORIES = json.loads('${categoriesJson}')
|
|
|
|
|
|
class QBitMock(BaseHTTPRequestHandler):
|
|
def _respond(self, code=200, body=b"Ok.", content_type="text/plain"):
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Set-Cookie", "SID=mock_session_id; Path=/")
|
|
self.end_headers()
|
|
self.wfile.write(body if isinstance(body, bytes) else body.encode())
|
|
|
|
def do_GET(self):
|
|
path = self.path.split("?")[0]
|
|
if path == "/api/v2/app/webapiVersion":
|
|
self._respond(body=b"2.9.3")
|
|
elif path == "/api/v2/app/version":
|
|
self._respond(body=b"v5.0.0")
|
|
elif path == "/api/v2/torrents/info":
|
|
self._respond(body=b"[]", content_type="application/json")
|
|
elif path == "/api/v2/torrents/categories":
|
|
body = json.dumps(CATEGORIES).encode()
|
|
self._respond(body=body, content_type="application/json")
|
|
elif path == "/api/v2/app/preferences":
|
|
body = json.dumps({"save_path": "/tmp"}).encode()
|
|
self._respond(body=body, content_type="application/json")
|
|
else:
|
|
self._respond()
|
|
|
|
def do_POST(self):
|
|
content_length = int(self.headers.get("Content-Length", 0))
|
|
body = self.rfile.read(content_length).decode()
|
|
path = urlparse(self.path).path
|
|
query = parse_qs(urlparse(self.path).query)
|
|
form = parse_qs(body)
|
|
params = {**query, **form}
|
|
if path == "/api/v2/torrents/createCategory":
|
|
name = params.get("category", [""])[0]
|
|
save_path = params.get("savePath", params.get("save_path", [""]))[0] or "/downloads"
|
|
if name:
|
|
CATEGORIES[name] = {"name": name, "savePath": save_path}
|
|
self._respond()
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
|
|
HTTPServer(("0.0.0.0", ${toString port}), QBitMock).serve_forever()
|
|
'';
|
|
in
|
|
{
|
|
description = "Mock qBittorrent API";
|
|
wantedBy = [ "multi-user.target" ];
|
|
inherit before;
|
|
serviceConfig = {
|
|
ExecStart = "${pkgs.python3}/bin/python3 ${mockScript}";
|
|
Type = "simple";
|
|
};
|
|
};
|
|
|
|
# Mock SABnzbd API.
|
|
mkMockSabnzbd =
|
|
{
|
|
port ? 6012,
|
|
}:
|
|
let
|
|
mockScript = pkgs.writeScript "mock-sabnzbd.py" ''
|
|
import json
|
|
from http.server import HTTPServer, BaseHTTPRequestHandler
|
|
|
|
class SabMock(BaseHTTPRequestHandler):
|
|
def _respond(self, code=200, body=b'{"config": {}}', content_type="application/json"):
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", content_type)
|
|
self.end_headers()
|
|
self.wfile.write(body if isinstance(body, bytes) else body.encode())
|
|
|
|
def do_GET(self):
|
|
if "mode=config" in self.path or "mode=version" in self.path:
|
|
self._respond(body=b'{"config": {"misc": {"api_key": "test"}}, "version": "4.0.0"}')
|
|
elif "mode=get_config" in self.path:
|
|
self._respond(body=b'{"config": {"misc": {"complete_dir": "/downloads/usenet", "pre_check": false}, "categories": [{"name": "tv", "order": 0, "pp": "", "script": "Default", "dir": "tv"}], "sorters": []}}')
|
|
else:
|
|
self._respond()
|
|
|
|
def do_POST(self):
|
|
self._respond()
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
HTTPServer(("0.0.0.0", ${toString port}), SabMock).serve_forever()
|
|
'';
|
|
in
|
|
{
|
|
description = "Mock SABnzbd API";
|
|
wantedBy = [ "multi-user.target" ];
|
|
serviceConfig = {
|
|
ExecStart = "${pkgs.python3}/bin/python3 ${mockScript}";
|
|
Type = "simple";
|
|
};
|
|
};
|
|
}
|