From a0f947adf945433aa0dd758141b5a53aebb7d608 Mon Sep 17 00:00:00 2001 From: amar-python <52807662+amar-python@users.noreply.github.com> Date: Thu, 16 Jul 2026 02:27:17 +0000 Subject: [PATCH] feat(data-integrity): robust CsvParser, run lifecycle state machine, shell hardening - Extract CSV parsing into services/csv_parser.py (CsvParser class) with explicit handling of zero-byte files, ragged rows, encoding fallback (UTF-8 BOM -> Latin-1 with warning) and SQL-injection-style headers; migration_service now delegates to it and logs every issue - Add READY and ERROR states to RunStatus with an explicit allowed- transition map, can_transition() and MigrationRun.transition_to() raising InvalidStateTransition on illegal moves - Flag runs as ERROR when an upload fails mid-way (with recovery path); validation now ends in READY (passed) or ERROR (failed) - Harden csv_loader.sh: strict table-name regex (blocks ';', '--', quotes, spaces) + 63-char limit; validate --env token in both csv_loader.sh and csv_utilise.sh; replace eval with indirect expansion in csv_utilise.sh - Frontend: status chip colors for new 'ready'/'error' states - Tests: test_csv_parser.py (28 cases) + test_run_lifecycle.py (39 cases) --- backend/database/models.py | 71 +++++++- backend/migration/build/csv_loader.sh | 21 +++ backend/migration/build/csv_utilise.sh | 13 +- backend/services/csv_parser.py | 222 +++++++++++++++++++++++++ backend/services/execution_service.py | 6 +- backend/services/migration_service.py | 122 ++++++++------ backend/services/schema_service.py | 10 +- backend/tests/test_csv_parser.py | 175 +++++++++++++++++++ backend/tests/test_run_lifecycle.py | 140 ++++++++++++++++ frontend/src/pages/Dashboard.tsx | 4 +- 10 files changed, 718 insertions(+), 66 deletions(-) create mode 100644 backend/services/csv_parser.py create mode 100644 backend/tests/test_csv_parser.py create mode 100644 backend/tests/test_run_lifecycle.py diff --git a/backend/database/models.py b/backend/database/models.py index 8be991e..833ad10 100644 --- a/backend/database/models.py +++ b/backend/database/models.py @@ -25,13 +25,68 @@ class Base(DeclarativeBase): class RunStatus(str, enum.Enum): - """Lifecycle states for a migration run.""" + """Lifecycle states for a migration run. + + State machine:: + + CREATED ──▶ UPLOADING ──▶ VALIDATING ──▶ READY ──▶ MIGRATING ──▶ COMPLETED + │ │ │ │ │ + └────────────┴─────────────┴────────────┴───────────┴──▶ ERROR / FAILED + + ``ERROR`` marks a run whose upload/validation broke mid-way (recoverable by + re-uploading or re-validating); ``FAILED`` marks a migration execution that + ran and did not succeed. + """ CREATED = "created" UPLOADING = "uploading" VALIDATING = "validating" + READY = "ready" MIGRATING = "migrating" COMPLETED = "completed" FAILED = "failed" + ERROR = "error" + + +class InvalidStateTransition(Exception): + """Raised when a migration run is moved to a state it cannot reach.""" + + +# Explicit allowed transitions. Same-state transitions are always allowed +# (idempotent updates, e.g. uploading a second file while UPLOADING). +ALLOWED_TRANSITIONS: dict[RunStatus, frozenset[RunStatus]] = { + RunStatus.CREATED: frozenset({ + RunStatus.UPLOADING, RunStatus.VALIDATING, RunStatus.MIGRATING, RunStatus.ERROR, + }), + RunStatus.UPLOADING: frozenset({ + RunStatus.VALIDATING, RunStatus.READY, RunStatus.MIGRATING, + RunStatus.ERROR, RunStatus.CREATED, + }), + RunStatus.VALIDATING: frozenset({ + RunStatus.READY, RunStatus.ERROR, RunStatus.CREATED, + }), + RunStatus.READY: frozenset({ + RunStatus.UPLOADING, RunStatus.VALIDATING, RunStatus.MIGRATING, RunStatus.ERROR, + }), + RunStatus.MIGRATING: frozenset({ + RunStatus.COMPLETED, RunStatus.FAILED, RunStatus.ERROR, RunStatus.CREATED, + }), + RunStatus.COMPLETED: frozenset({ + RunStatus.VALIDATING, RunStatus.MIGRATING, RunStatus.ERROR, + }), + RunStatus.FAILED: frozenset({ + RunStatus.UPLOADING, RunStatus.VALIDATING, RunStatus.MIGRATING, RunStatus.ERROR, + }), + RunStatus.ERROR: frozenset({ + RunStatus.CREATED, RunStatus.UPLOADING, RunStatus.VALIDATING, RunStatus.MIGRATING, + }), +} + + +def can_transition(current: RunStatus, target: RunStatus) -> bool: + """Return True when ``current`` → ``target`` is a legal state change.""" + if current == target: + return True + return target in ALLOWED_TRANSITIONS.get(current, frozenset()) class MigrationRun(Base): @@ -65,6 +120,20 @@ class MigrationRun(Base): "UploadedFile", back_populates="migration_run", cascade="all, delete-orphan" ) + def transition_to(self, target: "RunStatus") -> None: + """Move the run to ``target``, enforcing the lifecycle state machine. + + Raises ``InvalidStateTransition`` for illegal moves (e.g. COMPLETED → + UPLOADING) so bugs surface immediately instead of corrupting run state. + """ + current = self.status if isinstance(self.status, RunStatus) else RunStatus(self.status) + if not can_transition(current, target): + raise InvalidStateTransition( + f"Migration run {self.id}: illegal transition " + f"{current.value!r} → {target.value!r}" + ) + self.status = target + def __repr__(self) -> str: return f"" diff --git a/backend/migration/build/csv_loader.sh b/backend/migration/build/csv_loader.sh index ca9e590..a19f334 100644 --- a/backend/migration/build/csv_loader.sh +++ b/backend/migration/build/csv_loader.sh @@ -89,6 +89,13 @@ done [[ -z "$CSV_FILE" ]] && { error "No CSV file specified."; usage; } [[ ! -f "$CSV_FILE" ]] && { error "File not found: $CSV_FILE"; exit 1; } +# Validate the env token: it is interpolated into variable names by the engine +# adapters (some via eval), so it must never contain shell metacharacters. +if [[ ! "$TARGET_ENV" =~ ^[a-zA-Z_][a-zA-Z0-9_]*$ ]]; then + error "Invalid --env value: '${TARGET_ENV}'. Allowed: letters, digits, underscore." + exit 1 +fi + # ── Load configuration ──────────────────────────────────────────────────────── if [[ -f "$CONFIG_LOCAL" ]]; then source "$CONFIG_LOCAL" @@ -118,6 +125,20 @@ TABLE_NAME="${TABLE_NAME%.CSV}" # Sanitise: lowercase, replace spaces/hyphens with underscores TABLE_NAME="$(echo "$TABLE_NAME" | tr '[:upper:]' '[:lower:]' | tr ' -' '__')" +# ── Validate table identifier (blocks `;`, `--`, quotes, spaces, etc.) ─────── +# The table name is interpolated into generated SQL, so it must be a strict +# SQL identifier: letters/digits/underscore, not starting with a digit. +if [[ ! "$TABLE_NAME" =~ ^[a-zA-Z_][a-zA-Z0-9_]*$ ]]; then + error "Invalid table name derived from input: '${TABLE_NAME}'." + error "Allowed: letters, digits, underscore; must not start with a digit." + error "Rename the CSV file or pass a safe name with --table ." + exit 1 +fi +if [[ "${#TABLE_NAME}" -gt 63 ]]; then + error "Table name '${TABLE_NAME}' exceeds PostgreSQL's 63-character identifier limit." + exit 1 +fi + # ── Setup log directory and files ───────────────────────────────────────────── mkdir -p "$LOG_DIR" TIMESTAMP="$(date +%Y%m%d_%H%M%S)" diff --git a/backend/migration/build/csv_utilise.sh b/backend/migration/build/csv_utilise.sh index 7246b0e..6aa61b6 100755 --- a/backend/migration/build/csv_utilise.sh +++ b/backend/migration/build/csv_utilise.sh @@ -146,12 +146,21 @@ if [[ "$DB_ENGINE" != "postgresql" ]]; then fi # ── Resolve PostgreSQL connection details ──────────────────────────────────── +# Validate the env token first: it is used to build variable names, so it must +# never contain shell metacharacters (blocks `--env 'x; rm -rf ~'` style abuse). +if [[ ! "$TARGET_ENV" =~ ^[a-zA-Z_][a-zA-Z0-9_]*$ ]]; then + error "Invalid --env value: '${TARGET_ENV}'. Allowed: letters, digits, underscore." + exit 1 +fi E="${TARGET_ENV^^}" PG_HOST="${PGHOST:-${PG_HOST:-localhost}}" PG_PORT="${PGPORT:-${PG_PORT:-5432}}" PG_USER="${PGUSER:-${PG_SUPERUSER:-postgres}}" -DB_NAME="$(eval echo "\$PG_DB_${E}")" -SCHEMA="$(eval echo "\$PG_SCHEMA_${E}")" +# Indirect expansion instead of eval — no code execution path. +_db_var="PG_DB_${E}" +_schema_var="PG_SCHEMA_${E}" +DB_NAME="${!_db_var:-}" +SCHEMA="${!_schema_var:-}" if [[ -z "$DB_NAME" || -z "$SCHEMA" ]]; then error "Could not resolve database / schema for env '${TARGET_ENV}'." diff --git a/backend/services/csv_parser.py b/backend/services/csv_parser.py new file mode 100644 index 0000000..5db36bc --- /dev/null +++ b/backend/services/csv_parser.py @@ -0,0 +1,222 @@ +"""Robust CSV parsing for uploaded migration files. + +``CsvParser`` centralises every defensive check that used to be scattered (or +silently swallowed) in ``migration_service._parse_csv_metadata``: + + * zero-byte / header-only files + * encoding fallback (UTF-8 with BOM → Latin-1) with an explicit warning + * ragged rows (data rows whose column count differs from the header) + * suspicious header names that could smuggle SQL fragments into generated + DDL (``;``, ``--``, quotes, comment markers, DROP/DELETE/... keywords) + +The parser never raises for malformed *content* — it returns a +``CsvParseResult`` whose ``issues`` list explains everything found, so the +caller decides whether to reject the file or store it with warnings. +""" +from __future__ import annotations + +import csv +import io +import json +import re +from dataclasses import dataclass, field +from typing import Optional + +# Characters/sequences in a header that are never legitimate column names and +# are classic SQL-injection vectors when identifiers are interpolated in DDL. +_FORBIDDEN_HEADER_PATTERN = re.compile(r"""[;'"`\\]|--|/\*|\*/""") + +# Statement keywords that should never appear as a standalone word inside a +# column header (e.g. ``name; DROP TABLE users``). +_FORBIDDEN_KEYWORDS = re.compile( + r"\b(drop|delete|truncate|insert|update|alter|grant|revoke|exec|execute)\b", + re.IGNORECASE, +) + +# Maximum header length we accept — anything longer is suspicious and would +# exceed PostgreSQL's 63-byte identifier limit anyway. +MAX_HEADER_LENGTH = 128 + + +@dataclass +class CsvIssue: + """A single problem found while parsing a CSV file.""" + + severity: str # "error" | "warning" + check: str # machine-readable check name + message: str # human-readable explanation + + def as_dict(self) -> dict: + return {"severity": self.severity, "check": self.check, "message": self.message} + + +@dataclass +class CsvParseResult: + """Outcome of parsing one CSV payload.""" + + ok: bool = False + row_count: Optional[int] = None + column_count: Optional[int] = None + columns: Optional[list[str]] = None + encoding: Optional[str] = None + issues: list[CsvIssue] = field(default_factory=list) + + @property + def errors(self) -> list[CsvIssue]: + return [i for i in self.issues if i.severity == "error"] + + @property + def warnings(self) -> list[CsvIssue]: + return [i for i in self.issues if i.severity == "warning"] + + def columns_json(self) -> Optional[str]: + """Column names serialised as JSON, matching the legacy DB format.""" + return json.dumps(self.columns) if self.columns is not None else None + + +class CsvParser: + """Defensive CSV parser used by the upload pipeline.""" + + def __init__(self, max_ragged_examples: int = 5) -> None: + # How many ragged-row line numbers to include in the issue message. + self.max_ragged_examples = max_ragged_examples + + # -- decoding ---------------------------------------------------------- + + def _decode(self, content: bytes, result: CsvParseResult) -> Optional[str]: + """Decode bytes to text, preferring UTF-8 (BOM-aware), falling back + to Latin-1 with a warning. Returns None if undecodable.""" + try: + result.encoding = "utf-8" + return content.decode("utf-8-sig") + except UnicodeDecodeError: + pass + try: + text = content.decode("latin-1") + result.encoding = "latin-1" + result.issues.append(CsvIssue( + severity="warning", + check="encoding_fallback", + message="File is not valid UTF-8; decoded as Latin-1. " + "Non-ASCII characters may be misinterpreted.", + )) + return text + except UnicodeDecodeError: # pragma: no cover — latin-1 never fails + result.issues.append(CsvIssue( + severity="error", + check="undecodable", + message="File could not be decoded as UTF-8 or Latin-1.", + )) + return None + + # -- header checks ----------------------------------------------------- + + def _check_headers(self, headers: list[str], result: CsvParseResult) -> None: + for idx, header in enumerate(headers): + stripped = header.strip() + if not stripped: + result.issues.append(CsvIssue( + severity="warning", + check="empty_header", + message=f"Column {idx + 1} has an empty header name.", + )) + continue + if len(stripped) > MAX_HEADER_LENGTH: + result.issues.append(CsvIssue( + severity="error", + check="header_too_long", + message=f"Header {idx + 1} exceeds {MAX_HEADER_LENGTH} characters.", + )) + if _FORBIDDEN_HEADER_PATTERN.search(stripped) or _FORBIDDEN_KEYWORDS.search(stripped): + result.issues.append(CsvIssue( + severity="error", + check="suspicious_header", + message=( + f"Header {idx + 1} ({stripped[:64]!r}) contains characters or " + "SQL keywords that are not allowed in column names." + ), + )) + + # -- main entry point ---------------------------------------------------- + + def parse(self, content: bytes, filename: str = "upload.csv") -> CsvParseResult: + """Parse CSV bytes and return a fully-populated ``CsvParseResult``. + + Never raises for malformed content; every problem is reported in + ``result.issues`` and ``result.ok`` is False when any error exists. + """ + result = CsvParseResult() + + # 1. Zero-byte / whitespace-only guard + if not content or not content.strip(): + result.issues.append(CsvIssue( + severity="error", + check="empty_file", + message=f"File {filename!r} is empty (0 bytes of data).", + )) + return result + + # 2. Decode with fallback + text = self._decode(content, result) + if text is None: + return result + + # 3. Header row + reader = csv.reader(io.StringIO(text)) + try: + headers = next(reader, None) + except csv.Error as exc: + result.issues.append(CsvIssue( + severity="error", + check="malformed_csv", + message=f"CSV structure could not be parsed: {exc}", + )) + return result + + if not headers or all(not h.strip() for h in headers): + result.issues.append(CsvIssue( + severity="error", + check="missing_header", + message="No header row found.", + )) + return result + + result.columns = [h.strip() for h in headers] + result.column_count = len(headers) + self._check_headers(headers, result) + + # 4. Data rows — count and detect ragged rows + expected = len(headers) + row_count = 0 + ragged: list[int] = [] + try: + for line_no, row in enumerate(reader, start=2): + if not row: # skip completely blank lines + continue + row_count += 1 + if len(row) != expected: + ragged.append(line_no) + except csv.Error as exc: + result.issues.append(CsvIssue( + severity="error", + check="malformed_csv", + message=f"CSV structure broke mid-file: {exc}", + )) + return result + + result.row_count = row_count + if ragged: + sample = ", ".join(map(str, ragged[: self.max_ragged_examples])) + more = f" (+{len(ragged) - self.max_ragged_examples} more)" \ + if len(ragged) > self.max_ragged_examples else "" + result.issues.append(CsvIssue( + severity="error", + check="ragged_rows", + message=( + f"{len(ragged)} row(s) have a column count different from the " + f"header ({expected} expected). Lines: {sample}{more}." + ), + )) + + result.ok = not result.errors + return result diff --git a/backend/services/execution_service.py b/backend/services/execution_service.py index abf468b..3614632 100644 --- a/backend/services/execution_service.py +++ b/backend/services/execution_service.py @@ -100,7 +100,7 @@ def execute_migration(db: Session, run_id: int) -> Optional[dict[str, Any]]: if not run: return None - run.status = RunStatus.MIGRATING + run.transition_to(RunStatus.MIGRATING) db.commit() files = ( @@ -110,7 +110,7 @@ def execute_migration(db: Session, run_id: int) -> Optional[dict[str, Any]]: ) if not files: - run.status = RunStatus.CREATED + run.transition_to(RunStatus.CREATED) db.commit() return { "run_id": run_id, @@ -178,7 +178,7 @@ def execute_migration(db: Session, run_id: int) -> Optional[dict[str, Any]]: "rows_loaded": 0, }) - run.status = RunStatus.COMPLETED if not has_error else RunStatus.FAILED + run.transition_to(RunStatus.COMPLETED if not has_error else RunStatus.FAILED) db.commit() return { diff --git a/backend/services/migration_service.py b/backend/services/migration_service.py index 445a5a0..76c085d 100644 --- a/backend/services/migration_service.py +++ b/backend/services/migration_service.py @@ -3,9 +3,7 @@ Keeps route handlers thin: they delegate to functions here, which interact with the ORM models and the filesystem. """ -import csv -import io -import json +import logging import os import uuid from typing import Optional @@ -14,6 +12,9 @@ from sqlalchemy.orm import Session from database.models import MigrationRun, RunStatus, UploadedFile +from services.csv_parser import CsvParser + +logger = logging.getLogger("mep.migration_service") # Upload directory — relative to the backend working dir, mounted via Docker UPLOAD_DIR = os.environ.get("MEP_UPLOAD_DIR", os.path.join(os.getcwd(), "uploads")) @@ -82,69 +83,80 @@ def delete_run(db: Session, run_id: int) -> bool: # --------------------------------------------------------------------------- def _parse_csv_metadata(content: bytes, filename: str) -> dict: - """Quickly parse CSV content to extract row count, column count, and column names.""" - result = {"row_count": None, "column_count": None, "columns": None} - try: - text = content.decode("utf-8-sig") # handles BOM - reader = csv.reader(io.StringIO(text)) - headers = next(reader, None) - if headers: - result["columns"] = json.dumps([h.strip() for h in headers]) - result["column_count"] = len(headers) - # Count data rows (excluding header) - row_count = sum(1 for _ in reader) - result["row_count"] = row_count - except Exception: - pass # non-fatal — metadata is optional - return result + """Parse CSV content to extract row count, column count, and column names. + + Delegates to the robust ``CsvParser``; parse problems are logged (not + swallowed silently) and reflected as None metadata, which surfaces to the + UI as "unknown" rather than pretending the file is fine. + """ + parsed = CsvParser().parse(content, filename) + for issue in parsed.issues: + logger.warning("CSV %s [%s/%s]: %s", filename, issue.severity, issue.check, issue.message) + return { + "row_count": parsed.row_count, + "column_count": parsed.column_count, + "columns": parsed.columns_json(), + } async def upload_file(db: Session, run_id: int, file: UploadFile) -> Optional[UploadedFile]: """Save an uploaded file to disk and create a DB record. Returns the UploadedFile object, or None if the run doesn't exist. + If the upload fails mid-way (disk error, DB error, client disconnect), + the run is flagged as ERROR so it never lingers in UPLOADING. """ run = get_run(db, run_id) if not run: return None - # Read file content - content = await file.read() - file_size = len(content) - - # Generate a unique stored filename to avoid collisions - ext = os.path.splitext(file.filename or "file.csv")[1] - stored_name = f"{uuid.uuid4().hex}{ext}" - - # Save to disk - run_dir = _ensure_upload_dir(run_id) - file_path = os.path.join(run_dir, stored_name) - with open(file_path, "wb") as f: - f.write(content) - - # Parse CSV metadata - meta = _parse_csv_metadata(content, file.filename or "unknown.csv") - - # Persist to DB - uploaded = UploadedFile( - migration_run_id=run_id, - original_filename=file.filename or "unknown.csv", - stored_filename=stored_name, - file_size=file_size, - content_type=file.content_type or "text/csv", - row_count=meta["row_count"], - column_count=meta["column_count"], - columns=meta["columns"], - ) - db.add(uploaded) - - # Update run status to UPLOADING if still CREATED - if run.status == RunStatus.CREATED: - run.status = RunStatus.UPLOADING - - db.commit() - db.refresh(uploaded) - return uploaded + try: + # Read file content + content = await file.read() + file_size = len(content) + + # Generate a unique stored filename to avoid collisions + ext = os.path.splitext(file.filename or "file.csv")[1] + stored_name = f"{uuid.uuid4().hex}{ext}" + + # Save to disk + run_dir = _ensure_upload_dir(run_id) + file_path = os.path.join(run_dir, stored_name) + with open(file_path, "wb") as f: + f.write(content) + + # Parse CSV metadata + meta = _parse_csv_metadata(content, file.filename or "unknown.csv") + + # Persist to DB + uploaded = UploadedFile( + migration_run_id=run_id, + original_filename=file.filename or "unknown.csv", + stored_filename=stored_name, + file_size=file_size, + content_type=file.content_type or "text/csv", + row_count=meta["row_count"], + column_count=meta["column_count"], + columns=meta["columns"], + ) + db.add(uploaded) + + # Move the run (back) into UPLOADING for any state that allows it — + # a fresh run, a recovered ERROR/FAILED run, or a READY run receiving + # additional files (which must then be re-validated). + if run.status in (RunStatus.CREATED, RunStatus.ERROR, RunStatus.FAILED, RunStatus.READY): + run.transition_to(RunStatus.UPLOADING) + + db.commit() + db.refresh(uploaded) + return uploaded + except Exception: + # Flag the run as ERROR — an upload died mid-way. + logger.exception("Upload failed mid-way for run %s; flagging run as ERROR", run_id) + db.rollback() + run.transition_to(RunStatus.ERROR) + db.commit() + raise def list_files(db: Session, run_id: int) -> list[UploadedFile]: diff --git a/backend/services/schema_service.py b/backend/services/schema_service.py index e333556..14b66aa 100644 --- a/backend/services/schema_service.py +++ b/backend/services/schema_service.py @@ -244,7 +244,7 @@ def validate_run(db: Session, run_id: int) -> Optional[dict]: if not run: return None - run.status = RunStatus.VALIDATING + run.transition_to(RunStatus.VALIDATING) db.commit() files = ( @@ -254,7 +254,8 @@ def validate_run(db: Session, run_id: int) -> Optional[dict]: ) if not files: - run.status = RunStatus.CREATED + # Nothing to validate — revert to CREATED (pre-upload state) + run.transition_to(RunStatus.CREATED) db.commit() return { "run_id": run_id, @@ -300,8 +301,9 @@ def validate_run(db: Session, run_id: int) -> Optional[dict]: }) passed = total_errors == 0 - # Revert to CREATED status (validation is a pre-check, not a final state) - run.status = RunStatus.CREATED if passed else RunStatus.CREATED + # Lifecycle: validation passed → READY (eligible for execution); + # validation failed → ERROR (must fix files and re-validate). + run.transition_to(RunStatus.READY if passed else RunStatus.ERROR) db.commit() return { diff --git a/backend/tests/test_csv_parser.py b/backend/tests/test_csv_parser.py new file mode 100644 index 0000000..f385b5a --- /dev/null +++ b/backend/tests/test_csv_parser.py @@ -0,0 +1,175 @@ +"""Unit tests for services.csv_parser.CsvParser. + +Covers the failure modes the parser must handle defensively: +zero-byte files, ragged rows, encoding mismatches, and malicious +(SQL-injection-style) header names. +""" +import json + +import pytest + +from services.csv_parser import CsvParser + + +@pytest.fixture +def parser() -> CsvParser: + return CsvParser() + + +# --------------------------------------------------------------------------- +# Happy path +# --------------------------------------------------------------------------- + +class TestHappyPath: + + def test_simple_csv(self, parser): + result = parser.parse(b"id,name,email\n1,Alice,a@x.com\n2,Bob,b@x.com\n") + assert result.ok + assert result.row_count == 2 + assert result.column_count == 3 + assert result.columns == ["id", "name", "email"] + assert result.encoding == "utf-8" + assert result.issues == [] + + def test_utf8_bom_is_stripped(self, parser): + result = parser.parse("id,name\n1,Ünïcode\n".encode("utf-8-sig")) + assert result.ok + assert result.columns == ["id", "name"] + assert result.row_count == 1 + + def test_columns_json_matches_legacy_format(self, parser): + result = parser.parse(b"a,b\n1,2\n") + assert json.loads(result.columns_json()) == ["a", "b"] + + def test_blank_lines_are_skipped(self, parser): + result = parser.parse(b"a,b\n1,2\n\n\n3,4\n") + assert result.ok + assert result.row_count == 2 + + def test_quoted_fields_with_commas(self, parser): + result = parser.parse(b'id,note\n1,"hello, world"\n') + assert result.ok + assert result.row_count == 1 + + +# --------------------------------------------------------------------------- +# Zero-byte / empty files +# --------------------------------------------------------------------------- + +class TestEmptyFiles: + + def test_zero_byte_file(self, parser): + result = parser.parse(b"", "empty.csv") + assert not result.ok + assert result.row_count is None + assert any(i.check == "empty_file" for i in result.errors) + + def test_whitespace_only_file(self, parser): + result = parser.parse(b" \n \n", "blank.csv") + assert not result.ok + assert any(i.check == "empty_file" for i in result.errors) + + def test_header_only_file_is_ok_with_zero_rows(self, parser): + result = parser.parse(b"id,name\n") + assert result.ok + assert result.row_count == 0 + assert result.column_count == 2 + + +# --------------------------------------------------------------------------- +# Ragged rows (inconsistent column counts) +# --------------------------------------------------------------------------- + +class TestRaggedRows: + + def test_row_with_too_few_columns(self, parser): + result = parser.parse(b"a,b,c\n1,2,3\n1,2\n") + assert not result.ok + issue = next(i for i in result.errors if i.check == "ragged_rows") + assert "1 row(s)" in issue.message + assert "3" in issue.message # line number of the bad row + + def test_row_with_too_many_columns(self, parser): + result = parser.parse(b"a,b\n1,2\n1,2,3,4\n") + assert not result.ok + assert any(i.check == "ragged_rows" for i in result.errors) + + def test_multiple_ragged_rows_reported_with_line_numbers(self, parser): + content = b"a,b\n" + b"1\n" * 10 + result = parser.parse(content) + assert not result.ok + issue = next(i for i in result.errors if i.check == "ragged_rows") + assert "10 row(s)" in issue.message + assert "more" in issue.message # truncated example list + + def test_consistent_rows_have_no_ragged_issue(self, parser): + result = parser.parse(b"a,b\n1,2\n3,4\n") + assert not any(i.check == "ragged_rows" for i in result.issues) + + +# --------------------------------------------------------------------------- +# Encoding mismatches +# --------------------------------------------------------------------------- + +class TestEncodingMismatch: + + def test_latin1_falls_back_with_warning(self, parser): + content = "id,name\n1,Caf\xe9\n".encode("latin-1") # 0xE9 is invalid UTF-8 + result = parser.parse(content) + assert result.ok # decodable, so parse succeeds… + assert result.encoding == "latin-1" + assert any(i.check == "encoding_fallback" for i in result.warnings) + + def test_utf16_content_is_not_silently_accepted(self, parser): + content = "id,name\n1,Alice\n".encode("utf-16") + result = parser.parse(content) + # UTF-16 bytes decode as latin-1 garbage — the fallback warning must fire + assert any(i.check == "encoding_fallback" for i in result.warnings) + + def test_valid_utf8_has_no_encoding_warning(self, parser): + result = parser.parse("id,name\n1,Zoë\n".encode("utf-8")) + assert result.ok + assert result.encoding == "utf-8" + assert not result.warnings + + +# --------------------------------------------------------------------------- +# Malicious / SQL-injection-style headers +# --------------------------------------------------------------------------- + +class TestMaliciousHeaders: + + @pytest.mark.parametrize("header", [ + "name; DROP TABLE users", + "name;--", + "name'--", + 'name"', + "name/*comment*/", + "DROP", + "1; DELETE FROM runs", + "col`name", + "a\\b", + ]) + def test_injection_symbols_are_flagged(self, parser, header): + content = f"id,{header}\n1,x\n".encode("utf-8") + result = parser.parse(content) + assert not result.ok, f"header {header!r} should be rejected" + assert any(i.check == "suspicious_header" for i in result.errors) + + @pytest.mark.parametrize("header", [ + "name", "first_name", "e-mail", "Order Total", "année", "drop_zone_id", + ]) + def test_legitimate_headers_are_allowed(self, parser, header): + content = f"id,{header}\n1,x\n".encode("utf-8") + result = parser.parse(content) + assert result.ok, f"header {header!r} should be accepted" + + def test_overlong_header_is_rejected(self, parser): + long_header = "h" * 200 + result = parser.parse(f"id,{long_header}\n1,x\n".encode("utf-8")) + assert not result.ok + assert any(i.check == "header_too_long" for i in result.errors) + + def test_empty_header_produces_warning(self, parser): + result = parser.parse(b"id,,name\n1,2,3\n") + assert any(i.check == "empty_header" for i in result.warnings) diff --git a/backend/tests/test_run_lifecycle.py b/backend/tests/test_run_lifecycle.py new file mode 100644 index 0000000..b1cf731 --- /dev/null +++ b/backend/tests/test_run_lifecycle.py @@ -0,0 +1,140 @@ +"""Tests for the MigrationRun lifecycle state machine. + +Verifies the explicit transition rules (CREATED → UPLOADING → VALIDATING → +READY → MIGRATING → COMPLETED, with ERROR/FAILED escape hatches), that +illegal transitions raise, and that a mid-way upload failure flags the run +as ERROR. +""" +import io +from unittest.mock import patch + +import pytest + +from database.models import ( + InvalidStateTransition, + MigrationRun, + RunStatus, + can_transition, +) + + +# --------------------------------------------------------------------------- +# Pure transition-rule tests +# --------------------------------------------------------------------------- + +class TestTransitionRules: + + @pytest.mark.parametrize("current,target", [ + (RunStatus.CREATED, RunStatus.UPLOADING), + (RunStatus.UPLOADING, RunStatus.VALIDATING), + (RunStatus.VALIDATING, RunStatus.READY), + (RunStatus.READY, RunStatus.MIGRATING), + (RunStatus.MIGRATING, RunStatus.COMPLETED), + (RunStatus.MIGRATING, RunStatus.FAILED), + (RunStatus.VALIDATING, RunStatus.ERROR), + (RunStatus.UPLOADING, RunStatus.ERROR), + (RunStatus.ERROR, RunStatus.UPLOADING), # recover by re-uploading + (RunStatus.ERROR, RunStatus.VALIDATING), # recover by re-validating + (RunStatus.FAILED, RunStatus.MIGRATING), # retry execution + (RunStatus.READY, RunStatus.UPLOADING), # add more files + ]) + def test_legal_transitions(self, current, target): + assert can_transition(current, target) + + @pytest.mark.parametrize("current,target", [ + (RunStatus.COMPLETED, RunStatus.UPLOADING), + (RunStatus.COMPLETED, RunStatus.CREATED), + (RunStatus.COMPLETED, RunStatus.READY), + (RunStatus.CREATED, RunStatus.COMPLETED), + (RunStatus.CREATED, RunStatus.READY), + (RunStatus.CREATED, RunStatus.FAILED), + (RunStatus.VALIDATING, RunStatus.MIGRATING), + (RunStatus.VALIDATING, RunStatus.COMPLETED), + (RunStatus.ERROR, RunStatus.COMPLETED), + (RunStatus.ERROR, RunStatus.READY), + ]) + def test_illegal_transitions(self, current, target): + assert not can_transition(current, target) + + @pytest.mark.parametrize("state", list(RunStatus)) + def test_same_state_transition_is_always_legal(self, state): + assert can_transition(state, state) + + def test_every_state_is_covered_by_the_map(self): + from database.models import ALLOWED_TRANSITIONS + assert set(ALLOWED_TRANSITIONS) == set(RunStatus) + + +# --------------------------------------------------------------------------- +# transition_to() on the ORM model +# --------------------------------------------------------------------------- + +class TestTransitionTo: + + def test_transition_to_updates_status(self): + run = MigrationRun(name="t", status=RunStatus.CREATED) + run.transition_to(RunStatus.UPLOADING) + assert run.status == RunStatus.UPLOADING + + def test_illegal_transition_raises(self): + run = MigrationRun(name="t", status=RunStatus.COMPLETED) + with pytest.raises(InvalidStateTransition): + run.transition_to(RunStatus.UPLOADING) + + def test_illegal_transition_leaves_status_untouched(self): + run = MigrationRun(name="t", status=RunStatus.COMPLETED) + with pytest.raises(InvalidStateTransition): + run.transition_to(RunStatus.CREATED) + assert run.status == RunStatus.COMPLETED + + +# --------------------------------------------------------------------------- +# API-level lifecycle behaviour +# --------------------------------------------------------------------------- + +class TestLifecycleViaApi: + + def _create_run(self, client) -> int: + return client.post("/api/migrations", json={"name": "Lifecycle"}).json()["id"] + + def _upload(self, client, run_id, content=b"id,name\n1,Alice\n"): + return client.post( + f"/api/migrations/{run_id}/files", + files=[("files", ("data.csv", io.BytesIO(content), "text/csv"))], + ) + + def test_upload_moves_run_to_uploading(self, client): + run_id = self._create_run(client) + assert self._upload(client, run_id).status_code == 201 + assert client.get(f"/api/migrations/{run_id}").json()["status"] == "uploading" + + def test_successful_validation_moves_run_to_ready(self, client): + run_id = self._create_run(client) + self._upload(client, run_id) + resp = client.post(f"/api/migrations/{run_id}/validate") + assert resp.status_code == 200 + assert resp.json()["summary"]["passed"] is True + assert client.get(f"/api/migrations/{run_id}").json()["status"] == "ready" + + def test_failed_upload_flags_run_as_error(self, client): + """If the upload dies mid-way, the run must be flagged ERROR.""" + run_id = self._create_run(client) + with patch( + "services.migration_service._ensure_upload_dir", + side_effect=OSError("disk full"), + ): + with pytest.raises(OSError): + self._upload(client, run_id) + assert client.get(f"/api/migrations/{run_id}").json()["status"] == "error" + + def test_run_recovers_from_error_by_reuploading(self, client): + run_id = self._create_run(client) + with patch( + "services.migration_service._ensure_upload_dir", + side_effect=OSError("disk full"), + ): + with pytest.raises(OSError): + self._upload(client, run_id) + # Retry without the failure — run should move back to UPLOADING + assert self._upload(client, run_id).status_code == 201 + assert client.get(f"/api/migrations/{run_id}").json()["status"] == "uploading" diff --git a/frontend/src/pages/Dashboard.tsx b/frontend/src/pages/Dashboard.tsx index 250a436..b5e083a 100644 --- a/frontend/src/pages/Dashboard.tsx +++ b/frontend/src/pages/Dashboard.tsx @@ -26,7 +26,9 @@ import { formatFileSize, formatDate } from '../utils/format'; const statusColor = (s: string): 'default' | 'primary' | 'success' | 'error' | 'warning' | 'info' => { switch (s) { case 'completed': return 'success'; - case 'failed': return 'error'; + case 'ready': return 'primary'; + case 'failed': + case 'error': return 'error'; case 'migrating': case 'validating': return 'warning'; case 'uploading': return 'info';