arr-init/tests/lib/mocks.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";
      };
    };
}