"""SQL database and file transfer connectors (app/crud/sql_providers.py, app/crud/file_transfer.py, and the sql_query / sql_execute / ftp_transfer Task nodes). SQL runs for real against this suite's own PostgreSQL; the file server is an in-memory stand-in for SFTP/FTP.""" import fnmatch import posixpath import pytest from sqlalchemy.engine import make_url from app.core.config import settings from app.crud import file_transfer as ft from app.crud.file_transfer import RemoteFile from app.crud.sql_providers import SqlError, bind_named, check_statement, jsonable def test_named_placeholders_become_driver_parameters(): sql = "SELECT id::text, 'a:b 10%' AS lit -- :ignored\nFROM t WHERE a = :a AND b LIKE :b AND c = :a" assert bind_named(sql, {"a": 1, "b": "x"}, "numeric") == ( "SELECT id::text, 'a:b 10%' AS lit -- :ignored\nFROM t WHERE a = $1 AND b LIKE $2 AND c = $1", [1, "x"], ) text, args = bind_named(sql, {"a": 1, "b": "x", "unused": 2}, "pyformat") assert "'a:b 10%%'" in text and "a = %(a)s" in text and args == {"a": 1, "b": "x"} assert bind_named("SELECT :a FROM dual", {"a": 1}, "named") == ("SELECT :a FROM dual", {"a": 1}) with pytest.raises(SqlError, match=":b"): bind_named("SELECT :a, :b", {"a": 1}, "numeric") @pytest.mark.parametrize( "sql", [ "DELETE FROM t", "SELECT 1; DROP TABLE t", "WITH gone AS (DELETE FROM t RETURNING *) SELECT * FROM gone", "SELECT * INTO copy_of_t FROM t", ], ) def test_queries_cannot_write(sql): with pytest.raises(SqlError): check_statement(sql, read_only=True) def test_harmless_queries_pass_the_check(): check_statement("SELECT replace(name, 'a', 'b') FROM t WHERE note = 'drop table; now' ORDER BY id DESC;", read_only=True) check_statement("UPDATE t SET a = :a", read_only=False) def test_values_are_made_json_safe(): import datetime import decimal import uuid uid = uuid.uuid4() assert jsonable([decimal.Decimal("3"), decimal.Decimal("2.5"), datetime.date(2026, 1, 2), uid, b"hi"]) == [3, 2.5, "2026-01-02", str(uid), "aGk="] def _pg_config(**extra) -> dict: url = make_url(settings.database_url) return {"provider": "postgresql", "host": url.host, "port": str(url.port or 5432), "database": url.database, "username": url.username, "password": url.password or "", "ssl": "prefer", **extra} async def _connector(client, auth_headers, ctype, config): resp = await client.post("/connectors", headers=auth_headers, json={"name": f"Test {ctype}", "type": ctype, "configuration": config, "status": "enabled"}) assert resp.status_code == 201, resp.text return resp.json()["id"] async def _run(client, auth_headers, node, context): nodes = [{"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "n"}]}, {"key": "n", "label": "Node", **node}] task = await client.post("/tasks", headers=auth_headers, json={"name": "Connector test", "nodes": nodes}) assert task.status_code == 201, task.text resp = await client.post(f"/tasks/{task.json()['id']}/run", headers=auth_headers, json={"context": context, "allow_draft": True}) assert resp.status_code == 200, resp.text return resp.json() async def test_sql_query_against_a_real_postgres(client, auth_headers): connector_id = await _connector(client, auth_headers, "database", _pg_config()) result = await _run( client, auth_headers, { "type": "sql_query", "connector_id": connector_id, "sql": "SELECT g AS n, :label || '-' || g AS code, 'a:b 5%' AS lit FROM generate_series(1, :count::int) AS g ORDER BY g", "sql_max_rows": 3, "input_params": [{"name": "label", "type": "string", "source_field": "prefix"}, {"name": "count", "type": "int", "source_field": "howMany"}], "output_params": [{"name": "rows", "type": "list"}, {"name": "code", "type": "string"}, {"name": "truncated", "type": "string"}], }, {"prefix": "RTI", "howMany": 5}, ) assert result["ok"], result["trace"] ctx = result["context"] assert ctx["code"] == "RTI-1" and ctx["truncated"] == "yes" assert ctx["rows"] == [{"n": i, "code": f"RTI-{i}", "lit": "a:b 5%"} for i in (1, 2, 3)] test = await client.post(f"/connectors/{connector_id}/test", headers=auth_headers) assert test.json()["success"] is True and test.json()["detail"].startswith("Connected to PostgreSQL") async def test_writes_need_a_connector_that_allows_them(client, auth_headers): read_only = await _connector(client, auth_headers, "database", _pg_config()) refused = await _run(client, auth_headers, {"type": "sql_execute", "connector_id": read_only, "sql": "SELECT 1"}, {}) assert not refused["ok"] and "read-only" in refused["trace"][-1]["detail"] sneaky = await _run(client, auth_headers, {"type": "sql_query", "connector_id": read_only, "sql": "SELECT 1; DROP TABLE users"}, {}) assert not sneaky["ok"] and "one statement" in sneaky["trace"][-1]["detail"] writable = await _connector(client, auth_headers, "database", _pg_config(allow_writes="yes")) ran = await _run( client, auth_headers, {"type": "sql_execute", "connector_id": writable, "sql": "SELECT :v::text AS echoed", "input_params": [{"name": "v", "type": "string", "source_field": "v"}], "output_params": [{"name": "echoed", "type": "string"}]}, {"v": "hello"}, ) assert ran["ok"] and ran["context"]["echoed"] == "hello" class FakeServer: """In-memory stand-in for an SFTP/FTP session.""" def __init__(self, files: dict[str, bytes]): self.files = dict(files) self.fingerprint = "SHA256:test" def list(self, path): entries = {} for p in self.files: if posixpath.dirname(p) == path: entries[p] = RemoteFile(posixpath.basename(p), p, len(self.files[p]), None, False) elif p.startswith(path.rstrip("/") + "/"): sub = path.rstrip("/") + "/" + p[len(path.rstrip("/")) + 1 :].split("/")[0] entries[sub] = RemoteFile(posixpath.basename(sub), sub, None, None, True) return list(entries.values()) def is_dir(self, path): return any(p.startswith(path.rstrip("/") + "/") for p in self.files) def get(self, path): return self.files[path] def put(self, path, data): self.files[path] = data def rename(self, src, dst): self.files[dst] = self.files.pop(src) def delete(self, path): del self.files[path] def close(self): pass @pytest.fixture def fake_server(monkeypatch): server = FakeServer({"/in/a.csv": b"a,b\n1,2\n", "/in/b.csv": b"x", "/in/notes.txt": b"n", "/archive/.keep": b""}) monkeypatch.setattr(ft, "_open", lambda config: server) return server SFTP = {"provider": "sftp", "host": "sftp.test", "username": "bot", "password": "pw", "base_dir": "/in"} async def test_file_transfer_round_trip_through_tasks(client, auth_headers, fake_server): connector_id = await _connector(client, auth_headers, "file_transfer", SFTP) base = {"type": "ftp_transfer", "connector_id": connector_id} listed = await _run(client, auth_headers, {**base, "ftp_operation": "list", "ftp_pattern": "*.csv", "output_params": [{"name": "names", "type": "list"}]}, {}) assert listed["ok"] and listed["context"]["names"] == ["a.csv", "b.csv"] downloaded = await _run( client, auth_headers, {**base, "ftp_operation": "download", "ftp_pattern": "*.csv", "output_params": [{"name": "file_ids", "type": "list"}, {"name": "file_id", "type": "string"}]}, {}, ) assert downloaded["ok"], downloaded["trace"] file_id = downloaded["context"]["file_id"] assert len(downloaded["context"]["file_ids"]) == 2 content = await client.get(f"/uploads/{file_id}/download", headers=auth_headers) assert content.status_code == 200 and content.content == b"a,b\n1,2\n" uploaded = await _run( client, auth_headers, {**base, "ftp_operation": "upload", "remote_path": "/out", "ftp_target_path": "report-{{no}}.csv", "input_params": [{"name": "file_id", "type": "string", "source_field": "fid"}, {"name": "no", "type": "string", "source_field": "no"}]}, {"fid": file_id, "no": "7"}, ) assert uploaded["ok"], uploaded["trace"] assert fake_server.files["/out/report-7.csv"] == b"a,b\n1,2\n" moved = await _run(client, auth_headers, {**base, "ftp_operation": "move", "ftp_pattern": "*.csv", "ftp_target_path": "/archive/"}, {}) assert moved["ok"] and {"/archive/a.csv", "/archive/b.csv"} <= set(fake_server.files) and "/in/a.csv" not in fake_server.files deleted = await _run(client, auth_headers, {**base, "ftp_operation": "delete", "remote_path": "notes.txt"}, {}) assert deleted["ok"] and "/in/notes.txt" not in fake_server.files test = await client.post(f"/connectors/{connector_id}/test", headers=auth_headers) assert test.json()["success"] is True and "SHA256:test" in test.json()["detail"] def test_sftp_refuses_an_unpinned_host_key(monkeypatch): """A pinned fingerprint that doesn't match stops before any login.""" import paramiko class Key: def asbytes(self): return b"server-key" logins = [] class FakeTransport: def __init__(self, addr): pass def start_client(self, timeout=None): pass def get_remote_server_key(self): return Key() def auth_password(self, user, password): logins.append(user) def close(self): pass monkeypatch.setattr(paramiko, "Transport", FakeTransport) with pytest.raises(ft.FileTransferError, match="doesn't match"): ft._Sftp({**SFTP, "host_key_fingerprint": "SHA256:somethingElse"}) assert logins == [] assert ft.host_key_fingerprint(b"server-key").startswith("SHA256:") async def test_providers_are_listed(client, auth_headers): db = await client.get("/connector-providers/database", headers=auth_headers) files = await client.get("/connector-providers/file_transfer", headers=auth_headers) assert [p["id"] for p in db.json()] == ["postgresql", "mysql", "sqlserver", "oracle"] assert [p["id"] for p in files.json()] == ["sftp", "ftps", "ftp"] assert (await client.get("/connector-providers/nope", headers=auth_headers)).status_code == 404 def test_pattern_helper_matches_like_the_node(): assert fnmatch.fnmatch("INV_2026.xlsx", "INV_*.xlsx")