From cbcd968952c64c1bdd6c962c573d2a8c0af23709 Mon Sep 17 00:00:00 2001 From: Tom Brooks Date: Wed, 13 May 2026 15:01:04 +0100 Subject: [PATCH 1/7] feat: add new TaskPipeline Class --- digital_land/pipeline/__init__.py | 1 + digital_land/pipeline/task.py | 226 ++++++++++++++++++++++++++++++ tests/unit/test_task_pipeline.py | 153 ++++++++++++++++++++ 3 files changed, 380 insertions(+) create mode 100644 digital_land/pipeline/task.py create mode 100644 tests/unit/test_task_pipeline.py diff --git a/digital_land/pipeline/__init__.py b/digital_land/pipeline/__init__.py index 0936fc54c..200d7dbea 100644 --- a/digital_land/pipeline/__init__.py +++ b/digital_land/pipeline/__init__.py @@ -7,3 +7,4 @@ run_pipeline, EntityNumGen, ) +from .task import TaskPipeline, TaskPipelineConfig # noqa: F401 diff --git a/digital_land/pipeline/task.py b/digital_land/pipeline/task.py new file mode 100644 index 000000000..4f0834265 --- /dev/null +++ b/digital_land/pipeline/task.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +import json +import logging +from datetime import date +from pathlib import Path +from typing import List, Optional + +import polars as pl +from pydantic import ConfigDict, Field +from pydantic.dataclasses import dataclass + +logger = logging.getLogger(__name__) + +TASK_COLUMNS = [ + "dataset", + "organisation", + "endpoint", + "resource", + "details", + "severity", + "responsibility", + "task-source", + "entry-date", +] + +_EMPTY_SCHEMA = {col: pl.Utf8 for col in TASK_COLUMNS} + + +@dataclass(config=ConfigDict(extra="forbid")) +class TaskPipelineConfig: + dataset: str + organisation: str + endpoint: str + log_path: Optional[Path] = None + issue_path: Optional[Path] = None + output_path: Optional[Path] = None + severity_filter: List[str] = Field(default_factory=lambda: ["error"]) + responsibility_filter: List[str] = Field(default_factory=lambda: ["external"]) + entry_date: Optional[str] = None + + +class TaskPipeline: + """Generates a task CSV from collection logs and/or issue logs.""" + + def __init__(self, config: Optional[TaskPipelineConfig] = None): + self.config = config + + def run( + self, + dataset: Optional[str] = None, + organisation: Optional[str] = None, + endpoint: Optional[str] = None, + log_path: Optional[Path] = None, + issue_path: Optional[Path] = None, + output_path: Optional[Path] = None, + severity_filter: Optional[List[str]] = None, + responsibility_filter: Optional[List[str]] = None, + entry_date: Optional[str] = None, + ) -> List[dict]: + cfg = self.config + dataset = dataset or (cfg.dataset if cfg else None) + organisation = organisation or (cfg.organisation if cfg else None) + endpoint = endpoint or (cfg.endpoint if cfg else None) + log_path = log_path or (cfg.log_path if cfg else None) + issue_path = issue_path or (cfg.issue_path if cfg else None) + output_path = output_path or (cfg.output_path if cfg else None) + severity_filter = severity_filter or (cfg.severity_filter if cfg else ["error"]) + responsibility_filter = responsibility_filter or ( + cfg.responsibility_filter if cfg else ["external"] + ) + entry_date = entry_date or (cfg.entry_date if cfg else str(date.today())) + + frames = [] + + if log_path and Path(log_path).exists(): + log_tasks = _tasks_from_log(log_path, dataset, organisation, entry_date) + if not log_tasks.is_empty(): + frames.append(log_tasks) + + if issue_path and Path(issue_path).exists(): + issue_tasks = _tasks_from_issues( + issue_path, + organisation, + endpoint, + severity_filter, + responsibility_filter, + entry_date, + ) + if not issue_tasks.is_empty(): + frames.append(issue_tasks) + + result = pl.concat(frames) if frames else pl.DataFrame(schema=_EMPTY_SCHEMA) + tasks = result.to_dicts() + if output_path: + with open(output_path, "w") as f: + json.dump(tasks, f, indent=2) + return tasks + + +def _tasks_from_log( + log_path: Path, + dataset: str, + organisation: str, + entry_date: str, +) -> pl.DataFrame: + """Return a task row for each failed collection log entry.""" + + # infer_schema_length=0 tells polars to read all columns as strings rather than + # infer datatypes based on the first 100 rows, the default behaviour. + df = pl.read_csv(log_path, infer_schema_length=0, null_values=[""]) + + failed = df.filter(pl.col("status") != "200") + + if failed.is_empty(): + return pl.DataFrame(schema=_EMPTY_SCHEMA) + + n = len(failed) + + details_col = failed.select( + # pl.struct bundles multiple columns so map_elements can access both at once + pl.struct(["status", "exception"]) + .map_elements( + lambda row: json.dumps( + { + # status codes are converted to ints and nulls come in as empty strings + "status": ( + int(row["status"]) + if row["status"] and row["status"].isdigit() + else row["status"] + ), + "exception": row["exception"] or "", + } + ), + return_dtype=pl.Utf8, + ) + .alias("details") + ) + + return pl.DataFrame( + { + "dataset": pl.Series([dataset] * n, dtype=pl.Utf8), + "organisation": pl.Series([organisation] * n, dtype=pl.Utf8), + "endpoint": failed["endpoint"], + "resource": failed["resource"], + "details": details_col["details"], + "severity": pl.Series(["error"] * n, dtype=pl.Utf8), + "responsibility": pl.Series(["external"] * n, dtype=pl.Utf8), + "task-source": pl.Series(["log"] * n, dtype=pl.Utf8), + "entry-date": pl.Series([entry_date] * n, dtype=pl.Utf8), + } + ) + + +def _tasks_from_issues( + issue_path: Path, + organisation: str, + endpoint: str, + severity_filter: List[str], + responsibility_filter: List[str], + entry_date: str, +) -> pl.DataFrame: + """Return one task row per (issue-type, resource, field, dataset) group.""" + + df = pl.read_csv(issue_path, infer_schema_length=0, null_values=[""]) + cols = set(df.columns) + + if "severity" in cols: + df = df.filter(pl.col("severity").is_in(severity_filter)) + if "responsibility" in cols: + df = df.filter(pl.col("responsibility").is_in(responsibility_filter)) + + if df.is_empty(): + return pl.DataFrame(schema=_EMPTY_SCHEMA) + + for col in [ + "issue-type", + "resource", + "field", + "dataset", + "severity", + "responsibility", + ]: + if col not in df.columns: + df = df.with_columns(pl.lit("").alias(col)) + + # We are counting every row, so there might be duplicate entities across multiple data sources + # I can add in a split so it behaves more like the current version in submit repo but holding off + # until I understand the data better and if that split is required. + grouped = df.group_by(["issue-type", "resource", "field", "dataset"]).agg( + [ + pl.len().alias("count"), + pl.first("severity"), + pl.first("responsibility"), + ] + ) + + grouped = grouped.with_columns( + # pl.struct bundles multiple columns so map_elements can access both at once + pl.struct(["issue-type", "count", "field"]) + .map_elements( + lambda row: json.dumps( + { + "issue_type": row["issue-type"] or "", + "count": row["count"], + "field": row["field"] or "", + } + ), + return_dtype=pl.Utf8, + ) + .alias("details") + ) + + return grouped.select( + [ + pl.col("dataset"), + pl.lit(organisation).alias("organisation"), + pl.lit(endpoint).alias("endpoint"), + pl.col("resource"), + pl.col("details"), + pl.col("severity"), + pl.col("responsibility"), + pl.lit("issue").alias("task-source"), + pl.lit(entry_date).alias("entry-date"), + ] + ) diff --git a/tests/unit/test_task_pipeline.py b/tests/unit/test_task_pipeline.py new file mode 100644 index 000000000..9905599f1 --- /dev/null +++ b/tests/unit/test_task_pipeline.py @@ -0,0 +1,153 @@ +import json +import pytest + +from digital_land.pipeline.task import TaskPipeline, TaskPipelineConfig + + +@pytest.fixture +def log_csv(tmp_path): + """A collection log with one success and two failures.""" + path = tmp_path / "log.csv" + path.write_text( + "endpoint,resource,status,exception,entry-date,bytes,elapsed\n" + "endpoint-aaa,resource-aaa,200,,2024-01-01,1234,0.5\n" + "endpoint-bbb,resource-bbb,404,,2024-01-01,0,0.1\n" + "endpoint-ccc,,500,Connection refused,2024-01-01,0,0.0\n" + ) + return path + + +@pytest.fixture +def issue_csv(tmp_path): + """An issue log with error/external issues and one warning to be filtered out.""" + path = tmp_path / "issue.csv" + path.write_text( + "dataset,resource,line-number,entry-number,field,entity,issue-type,value,message,severity,responsibility\n" + "tree-preservation-zone,resource-aaa,1,1,geometry,,invalid-geometry,bad-wkt,Invalid geometry,error,external\n" + "tree-preservation-zone,resource-aaa,2,2,geometry,,invalid-geometry,other-bad-wkt,Invalid geometry,error,external\n" + "tree-preservation-zone,resource-aaa,3,3,name,,missing-value,,Missing value,error,external\n" + "tree-preservation-zone,resource-aaa,4,4,notes,,missing-value,,Missing value,warning,internal\n" + ) + return path + + +def test_log_tasks_creates_row_per_failure(log_csv): + pipeline = TaskPipeline() + results = pipeline.run( + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + log_path=log_csv, + ) + + # Only the two failed rows should produce tasks, not the 200 + assert len(results) == 2 + assert set(result["task-source"] for result in results) == {"log"} + + +def test_log_task_details_json(log_csv): + results = TaskPipeline().run( + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + log_path=log_csv, + ) + + details = [json.loads(result["details"]) for result in results] + statuses = {detail["status"] for detail in details} + assert 404 in statuses + assert 500 in statuses + # status should be an int in the JSON, not a string + assert all(isinstance(d["status"], int) for d in details) + + +def test_log_tasks_empty_when_all_successful(tmp_path): + log = tmp_path / "log.csv" + log.write_text( + "endpoint,resource,status,exception,entry-date,bytes,elapsed\n" + "endpoint-aaa,resource-aaa,200,,2024-01-01,1234,0.5\n" + ) + results = TaskPipeline().run( + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + log_path=log, + ) + assert results == [] + + +def test_issue_tasks_groups_by_type_and_field(issue_csv): + results = TaskPipeline().run( + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + issue_path=issue_csv, + ) + + # 2 geometry rows + 1 name row = 2 groups (invalid-geometry/geometry and missing-value/name) + # the warning/internal row should be filtered out + assert len(results) == 2 + assert set(result["task-source"] for result in results) == {"issue"} + + +def test_issue_task_details_json(issue_csv): + results = TaskPipeline().run( + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + issue_path=issue_csv, + ) + + details = [json.loads(result["details"]) for result in results] + geom_task = next( + detail for detail in details if detail["issue_type"] == "invalid-geometry" + ) + assert geom_task["count"] == 2 + assert geom_task["field"] == "geometry" + + +def test_both_sources_combined(log_csv, issue_csv): + results = TaskPipeline().run( + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + log_path=log_csv, + issue_path=issue_csv, + ) + + sources = set(result["task-source"] for result in results) + + assert "log" in sources + assert "issue" in sources + + +def test_run_with_config(log_csv): + config = TaskPipelineConfig( + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + log_path=log_csv, + ) + results = TaskPipeline(config).run() + assert len(results) == 2 + + +def test_output_has_correct_columns(log_csv): + results = TaskPipeline().run( + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + log_path=log_csv, + ) + expected_cols = { + "dataset", + "organisation", + "endpoint", + "resource", + "details", + "severity", + "responsibility", + "task-source", + "entry-date", + } + assert set(results[0].keys()) == expected_cols From 90b0f7c6fe869ab4146073b2c7a5c5c542bfa5ea Mon Sep 17 00:00:00 2001 From: Tom Brooks Date: Thu, 14 May 2026 14:18:03 +0100 Subject: [PATCH 2/7] feat: produce csv at output_path and status Enum --- digital_land/pipeline/__init__.py | 2 +- digital_land/pipeline/task.py | 100 ++++++++++++++++++------------ tests/unit/test_task_pipeline.py | 99 ++++++++++++++++++----------- 3 files changed, 125 insertions(+), 76 deletions(-) diff --git a/digital_land/pipeline/__init__.py b/digital_land/pipeline/__init__.py index 200d7dbea..61845963d 100644 --- a/digital_land/pipeline/__init__.py +++ b/digital_land/pipeline/__init__.py @@ -7,4 +7,4 @@ run_pipeline, EntityNumGen, ) -from .task import TaskPipeline, TaskPipelineConfig # noqa: F401 +from .task import TaskPipeline, TaskPipelineConfig, TaskPipelineStatus # noqa: F401 diff --git a/digital_land/pipeline/task.py b/digital_land/pipeline/task.py index 4f0834265..9811d321f 100644 --- a/digital_land/pipeline/task.py +++ b/digital_land/pipeline/task.py @@ -3,6 +3,7 @@ import json import logging from datetime import date +from enum import Enum from pathlib import Path from typing import List, Optional @@ -27,14 +28,20 @@ _EMPTY_SCHEMA = {col: pl.Utf8 for col in TASK_COLUMNS} +class TaskPipelineStatus(Enum): + SUCCESS = 1 + NO_TASKS = 2 + FAILED = 3 + + @dataclass(config=ConfigDict(extra="forbid")) class TaskPipelineConfig: dataset: str organisation: str endpoint: str + output_path: Path log_path: Optional[Path] = None issue_path: Optional[Path] = None - output_path: Optional[Path] = None severity_filter: List[str] = Field(default_factory=lambda: ["error"]) responsibility_filter: List[str] = Field(default_factory=lambda: ["external"]) entry_date: Optional[str] = None @@ -48,54 +55,67 @@ def __init__(self, config: Optional[TaskPipelineConfig] = None): def run( self, + output_path: Path = None, dataset: Optional[str] = None, organisation: Optional[str] = None, endpoint: Optional[str] = None, log_path: Optional[Path] = None, issue_path: Optional[Path] = None, - output_path: Optional[Path] = None, severity_filter: Optional[List[str]] = None, responsibility_filter: Optional[List[str]] = None, entry_date: Optional[str] = None, - ) -> List[dict]: - cfg = self.config - dataset = dataset or (cfg.dataset if cfg else None) - organisation = organisation or (cfg.organisation if cfg else None) - endpoint = endpoint or (cfg.endpoint if cfg else None) - log_path = log_path or (cfg.log_path if cfg else None) - issue_path = issue_path or (cfg.issue_path if cfg else None) - output_path = output_path or (cfg.output_path if cfg else None) - severity_filter = severity_filter or (cfg.severity_filter if cfg else ["error"]) - responsibility_filter = responsibility_filter or ( - cfg.responsibility_filter if cfg else ["external"] - ) - entry_date = entry_date or (cfg.entry_date if cfg else str(date.today())) - - frames = [] - - if log_path and Path(log_path).exists(): - log_tasks = _tasks_from_log(log_path, dataset, organisation, entry_date) - if not log_tasks.is_empty(): - frames.append(log_tasks) - - if issue_path and Path(issue_path).exists(): - issue_tasks = _tasks_from_issues( - issue_path, - organisation, - endpoint, - severity_filter, - responsibility_filter, - entry_date, + ) -> TaskPipelineStatus: + try: + cfg = self.config + output_path = output_path or (cfg.output_path if cfg else None) + dataset = dataset or (cfg.dataset if cfg else None) + organisation = organisation or (cfg.organisation if cfg else None) + endpoint = endpoint or (cfg.endpoint if cfg else None) + log_path = log_path or (cfg.log_path if cfg else None) + issue_path = issue_path or (cfg.issue_path if cfg else None) + severity_filter = severity_filter or ( + cfg.severity_filter if cfg else ["error"] ) - if not issue_tasks.is_empty(): - frames.append(issue_tasks) - - result = pl.concat(frames) if frames else pl.DataFrame(schema=_EMPTY_SCHEMA) - tasks = result.to_dicts() - if output_path: - with open(output_path, "w") as f: - json.dump(tasks, f, indent=2) - return tasks + responsibility_filter = responsibility_filter or ( + cfg.responsibility_filter if cfg else ["external"] + ) + entry_date = entry_date or (cfg.entry_date if cfg else str(date.today())) + + if output_path is None: + logger.error("output_path is required") + return TaskPipelineStatus.FAILED + + frames = [] + + if log_path and Path(log_path).exists(): + log_tasks = _tasks_from_log(log_path, dataset, organisation, entry_date) + if not log_tasks.is_empty(): + frames.append(log_tasks) + + if issue_path and Path(issue_path).exists(): + issue_tasks = _tasks_from_issues( + issue_path, + organisation, + endpoint, + severity_filter, + responsibility_filter, + entry_date, + ) + if not issue_tasks.is_empty(): + frames.append(issue_tasks) + + result = pl.concat(frames) if frames else pl.DataFrame(schema=_EMPTY_SCHEMA) + result.write_csv(output_path) + + return ( + TaskPipelineStatus.NO_TASKS + if result.is_empty() + else TaskPipelineStatus.SUCCESS + ) + + except Exception: + logger.exception("TaskPipeline failed") + return TaskPipelineStatus.FAILED def _tasks_from_log( diff --git a/tests/unit/test_task_pipeline.py b/tests/unit/test_task_pipeline.py index 9905599f1..f65ac2c80 100644 --- a/tests/unit/test_task_pipeline.py +++ b/tests/unit/test_task_pipeline.py @@ -1,7 +1,12 @@ +import csv import json import pytest -from digital_land.pipeline.task import TaskPipeline, TaskPipelineConfig +from digital_land.pipeline.task import ( + TaskPipeline, + TaskPipelineConfig, + TaskPipelineStatus, +) @pytest.fixture @@ -31,33 +36,42 @@ def issue_csv(tmp_path): return path -def test_log_tasks_creates_row_per_failure(log_csv): - pipeline = TaskPipeline() - results = pipeline.run( +def _read_csv(path): + with open(path) as f: + return list(csv.DictReader(f)) + + +def test_log_tasks_creates_row_per_failure(log_csv, tmp_path): + output = tmp_path / "tasks.csv" + status = TaskPipeline().run( + output_path=output, dataset="tree-preservation-zone", organisation="local-authority-eng:ABC", endpoint="endpoint-aaa", log_path=log_csv, ) - # Only the two failed rows should produce tasks, not the 200 - assert len(results) == 2 - assert set(result["task-source"] for result in results) == {"log"} + assert status == TaskPipelineStatus.SUCCESS + rows = _read_csv(output) + assert len(rows) == 2 + assert set(row["task-source"] for row in rows) == {"log"} -def test_log_task_details_json(log_csv): - results = TaskPipeline().run( +def test_log_task_details_json(log_csv, tmp_path): + output = tmp_path / "tasks.csv" + TaskPipeline().run( + output_path=output, dataset="tree-preservation-zone", organisation="local-authority-eng:ABC", endpoint="endpoint-aaa", log_path=log_csv, ) - details = [json.loads(result["details"]) for result in results] + rows = _read_csv(output) + details = [json.loads(row["details"]) for row in rows] statuses = {detail["status"] for detail in details} assert 404 in statuses assert 500 in statuses - # status should be an int in the JSON, not a string assert all(isinstance(d["status"], int) for d in details) @@ -67,47 +81,55 @@ def test_log_tasks_empty_when_all_successful(tmp_path): "endpoint,resource,status,exception,entry-date,bytes,elapsed\n" "endpoint-aaa,resource-aaa,200,,2024-01-01,1234,0.5\n" ) - results = TaskPipeline().run( + output = tmp_path / "tasks.csv" + status = TaskPipeline().run( + output_path=output, dataset="tree-preservation-zone", organisation="local-authority-eng:ABC", endpoint="endpoint-aaa", log_path=log, ) - assert results == [] + assert status == TaskPipelineStatus.NO_TASKS + assert _read_csv(output) == [] -def test_issue_tasks_groups_by_type_and_field(issue_csv): - results = TaskPipeline().run( +def test_issue_tasks_groups_by_type_and_field(issue_csv, tmp_path): + output = tmp_path / "tasks.csv" + status = TaskPipeline().run( + output_path=output, dataset="tree-preservation-zone", organisation="local-authority-eng:ABC", endpoint="endpoint-aaa", issue_path=issue_csv, ) - # 2 geometry rows + 1 name row = 2 groups (invalid-geometry/geometry and missing-value/name) - # the warning/internal row should be filtered out - assert len(results) == 2 - assert set(result["task-source"] for result in results) == {"issue"} + assert status == TaskPipelineStatus.SUCCESS + rows = _read_csv(output) + assert len(rows) == 2 + assert set(row["task-source"] for row in rows) == {"issue"} -def test_issue_task_details_json(issue_csv): - results = TaskPipeline().run( +def test_issue_task_details_json(issue_csv, tmp_path): + output = tmp_path / "tasks.csv" + TaskPipeline().run( + output_path=output, dataset="tree-preservation-zone", organisation="local-authority-eng:ABC", endpoint="endpoint-aaa", issue_path=issue_csv, ) - details = [json.loads(result["details"]) for result in results] - geom_task = next( - detail for detail in details if detail["issue_type"] == "invalid-geometry" - ) + rows = _read_csv(output) + details = [json.loads(row["details"]) for row in rows] + geom_task = next(d for d in details if d["issue_type"] == "invalid-geometry") assert geom_task["count"] == 2 assert geom_task["field"] == "geometry" -def test_both_sources_combined(log_csv, issue_csv): - results = TaskPipeline().run( +def test_both_sources_combined(log_csv, issue_csv, tmp_path): + output = tmp_path / "tasks.csv" + status = TaskPipeline().run( + output_path=output, dataset="tree-preservation-zone", organisation="local-authority-eng:ABC", endpoint="endpoint-aaa", @@ -115,25 +137,31 @@ def test_both_sources_combined(log_csv, issue_csv): issue_path=issue_csv, ) - sources = set(result["task-source"] for result in results) - + assert status == TaskPipelineStatus.SUCCESS + rows = _read_csv(output) + sources = set(row["task-source"] for row in rows) assert "log" in sources assert "issue" in sources -def test_run_with_config(log_csv): +def test_run_with_config(log_csv, tmp_path): + output = tmp_path / "tasks.csv" config = TaskPipelineConfig( dataset="tree-preservation-zone", organisation="local-authority-eng:ABC", endpoint="endpoint-aaa", + output_path=output, log_path=log_csv, ) - results = TaskPipeline(config).run() - assert len(results) == 2 + status = TaskPipeline(config).run() + assert status == TaskPipelineStatus.SUCCESS + assert len(_read_csv(output)) == 2 -def test_output_has_correct_columns(log_csv): - results = TaskPipeline().run( +def test_output_has_correct_columns(log_csv, tmp_path): + output = tmp_path / "tasks.csv" + TaskPipeline().run( + output_path=output, dataset="tree-preservation-zone", organisation="local-authority-eng:ABC", endpoint="endpoint-aaa", @@ -150,4 +178,5 @@ def test_output_has_correct_columns(log_csv): "task-source", "entry-date", } - assert set(results[0].keys()) == expected_cols + rows = _read_csv(output) + assert set(rows[0].keys()) == expected_cols From 8eeceadadbb83daa63293292c79b9aab115e6b11 Mon Sep 17 00:00:00 2001 From: Tom Brooks Date: Thu, 14 May 2026 14:50:59 +0100 Subject: [PATCH 3/7] adding in laze evaluation for polas and a task-id hash --- digital_land/pipeline/task.py | 65 +++++++++++++++++++++++++++----- tests/unit/test_task_pipeline.py | 1 + 2 files changed, 57 insertions(+), 9 deletions(-) diff --git a/digital_land/pipeline/task.py b/digital_land/pipeline/task.py index 9811d321f..382e0f2a9 100644 --- a/digital_land/pipeline/task.py +++ b/digital_land/pipeline/task.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib import json import logging from datetime import date @@ -23,6 +24,7 @@ "responsibility", "task-source", "entry-date", + "task-id", ] _EMPTY_SCHEMA = {col: pl.Utf8 for col in TASK_COLUMNS} @@ -126,11 +128,14 @@ def _tasks_from_log( ) -> pl.DataFrame: """Return a task row for each failed collection log entry.""" + # scan_csv + filter stays lazy; .collect() materialises only the failed rows # infer_schema_length=0 tells polars to read all columns as strings rather than # infer datatypes based on the first 100 rows, the default behaviour. - df = pl.read_csv(log_path, infer_schema_length=0, null_values=[""]) - - failed = df.filter(pl.col("status") != "200") + failed = ( + pl.scan_csv(log_path, infer_schema_length=0, null_values=[""]) + .filter(pl.col("status") != "200") + .collect() + ) if failed.is_empty(): return pl.DataFrame(schema=_EMPTY_SCHEMA) @@ -157,7 +162,7 @@ def _tasks_from_log( .alias("details") ) - return pl.DataFrame( + result = pl.DataFrame( { "dataset": pl.Series([dataset] * n, dtype=pl.Utf8), "organisation": pl.Series([organisation] * n, dtype=pl.Utf8), @@ -170,6 +175,26 @@ def _tasks_from_log( "entry-date": pl.Series([entry_date] * n, dtype=pl.Utf8), } ) + # callculate a sha256 hash from some key columns to identify this task + result = result.with_columns( + pl.struct(["dataset", "endpoint", "resource", "task-source", "details"]) + .map_elements( + lambda r: hashlib.sha256( + "|".join( + [ + r["dataset"] or "", + r["endpoint"] or "", + r["resource"] or "", + r["task-source"], + r["details"], + ] + ).encode() + ).hexdigest()[:16], + return_dtype=pl.Utf8, + ) + .alias("task-id") + ) + return result def _tasks_from_issues( @@ -182,13 +207,15 @@ def _tasks_from_issues( ) -> pl.DataFrame: """Return one task row per (issue-type, resource, field, dataset) group.""" - df = pl.read_csv(issue_path, infer_schema_length=0, null_values=[""]) - cols = set(df.columns) + lf = pl.scan_csv(issue_path, infer_schema_length=0, null_values=[""]) + cols = set(lf.columns) # .columns works on LazyFrame without collecting if "severity" in cols: - df = df.filter(pl.col("severity").is_in(severity_filter)) + lf = lf.filter(pl.col("severity").is_in(severity_filter)) if "responsibility" in cols: - df = df.filter(pl.col("responsibility").is_in(responsibility_filter)) + lf = lf.filter(pl.col("responsibility").is_in(responsibility_filter)) + + df = lf.collect() # materialise after filters are applied if df.is_empty(): return pl.DataFrame(schema=_EMPTY_SCHEMA) @@ -231,7 +258,7 @@ def _tasks_from_issues( .alias("details") ) - return grouped.select( + result = grouped.select( [ pl.col("dataset"), pl.lit(organisation).alias("organisation"), @@ -244,3 +271,23 @@ def _tasks_from_issues( pl.lit(entry_date).alias("entry-date"), ] ) + # callculate a sha256 hash from some key columns to identify this task + result = result.with_columns( + pl.struct(["dataset", "endpoint", "resource", "task-source", "details"]) + .map_elements( + lambda r: hashlib.sha256( + "|".join( + [ + r["dataset"] or "", + r["endpoint"] or "", + r["resource"] or "", + r["task-source"], + r["details"], + ] + ).encode() + ).hexdigest()[:16], + return_dtype=pl.Utf8, + ) + .alias("task-id") + ) + return result diff --git a/tests/unit/test_task_pipeline.py b/tests/unit/test_task_pipeline.py index f65ac2c80..079430e72 100644 --- a/tests/unit/test_task_pipeline.py +++ b/tests/unit/test_task_pipeline.py @@ -177,6 +177,7 @@ def test_output_has_correct_columns(log_csv, tmp_path): "responsibility", "task-source", "entry-date", + "task-id", } rows = _read_csv(output) assert set(rows[0].keys()) == expected_cols From 2a5c77f6a6080de28477e37f7cfed96ac29ed791 Mon Sep 17 00:00:00 2001 From: Tom Brooks Date: Fri, 15 May 2026 13:21:49 +0100 Subject: [PATCH 4/7] move tests to unit/pipeline/test_task.py --- tests/unit/pipeline/__init__.py | 0 tests/unit/pipeline/test_task.py | 178 ++++++++++++++++++++++++++++++ tests/unit/test_task_pipeline.py | 183 ------------------------------- 3 files changed, 178 insertions(+), 183 deletions(-) create mode 100644 tests/unit/pipeline/__init__.py create mode 100644 tests/unit/pipeline/test_task.py delete mode 100644 tests/unit/test_task_pipeline.py diff --git a/tests/unit/pipeline/__init__.py b/tests/unit/pipeline/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/unit/pipeline/test_task.py b/tests/unit/pipeline/test_task.py new file mode 100644 index 000000000..a969b74ad --- /dev/null +++ b/tests/unit/pipeline/test_task.py @@ -0,0 +1,178 @@ +import csv +import json +import pytest + +from digital_land.pipeline.task import ( + TaskPipeline, + TaskPipelineConfig, + TaskPipelineStatus, +) + + +@pytest.fixture +def log_csv(tmp_path): + """A collection log with one success and two failures.""" + path = tmp_path / "log.csv" + path.write_text( + "endpoint,resource,status,exception,entry-date,bytes,elapsed\n" + "endpoint-aaa,resource-aaa,200,,2024-01-01,1234,0.5\n" + "endpoint-bbb,resource-bbb,404,,2024-01-01,0,0.1\n" + "endpoint-ccc,,500,Connection refused,2024-01-01,0,0.0\n" + ) + return path + + +@pytest.fixture +def issue_csv(tmp_path): + """An issue log with error/external issues and one warning to be filtered out.""" + path = tmp_path / "issue.csv" + path.write_text( + "dataset,resource,line-number,entry-number,field,entity,issue-type,value,message,severity,responsibility\n" + "tree-preservation-zone,resource-aaa,1,1,geometry,,invalid-geometry,bad-wkt,Invalid geometry,error,external\n" + "tree-preservation-zone,resource-aaa,2,2,geometry,,invalid-geometry,other-bad-wkt,Invalid geometry,error,external\n" + "tree-preservation-zone,resource-aaa,3,3,name,,missing-value,,Missing value,error,external\n" + "tree-preservation-zone,resource-aaa,4,4,notes,,missing-value,,Missing value,warning,internal\n" + ) + return path + + +def _read_csv(path): + with open(path) as f: + return list(csv.DictReader(f)) + + +class TestTaskPipeline: + + def test_run_log_tasks_creates_row_per_failure(self, log_csv, tmp_path): + output = tmp_path / "tasks.csv" + status = TaskPipeline().run( + output_path=output, + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + log_path=log_csv, + ) + + assert status == TaskPipelineStatus.SUCCESS + rows = _read_csv(output) + assert len(rows) == 2 + assert set(row["task-source"] for row in rows) == {"log"} + + def test_run_log_task_details_json(self, log_csv, tmp_path): + output = tmp_path / "tasks.csv" + TaskPipeline().run( + output_path=output, + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + log_path=log_csv, + ) + + rows = _read_csv(output) + details = [json.loads(row["details"]) for row in rows] + statuses = {detail["status"] for detail in details} + assert 404 in statuses + assert 500 in statuses + assert all(isinstance(d["status"], int) for d in details) + + def test_run_log_tasks_empty_when_all_successful(self, tmp_path): + log = tmp_path / "log.csv" + log.write_text( + "endpoint,resource,status,exception,entry-date,bytes,elapsed\n" + "endpoint-aaa,resource-aaa,200,,2024-01-01,1234,0.5\n" + ) + output = tmp_path / "tasks.csv" + status = TaskPipeline().run( + output_path=output, + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + log_path=log, + ) + assert status == TaskPipelineStatus.NO_TASKS + assert _read_csv(output) == [] + + def test_run_issue_tasks_groups_by_type_and_field(self, issue_csv, tmp_path): + output = tmp_path / "tasks.csv" + status = TaskPipeline().run( + output_path=output, + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + issue_path=issue_csv, + ) + + assert status == TaskPipelineStatus.SUCCESS + rows = _read_csv(output) + assert len(rows) == 2 + assert set(row["task-source"] for row in rows) == {"issue"} + + def test_run_issue_task_details_json(self, issue_csv, tmp_path): + output = tmp_path / "tasks.csv" + TaskPipeline().run( + output_path=output, + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + issue_path=issue_csv, + ) + + rows = _read_csv(output) + details = [json.loads(row["details"]) for row in rows] + geom_task = next(d for d in details if d["issue_type"] == "invalid-geometry") + assert geom_task["count"] == 2 + assert geom_task["field"] == "geometry" + + def test_run_both_sources_combined(self, log_csv, issue_csv, tmp_path): + output = tmp_path / "tasks.csv" + status = TaskPipeline().run( + output_path=output, + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + log_path=log_csv, + issue_path=issue_csv, + ) + + assert status == TaskPipelineStatus.SUCCESS + rows = _read_csv(output) + sources = set(row["task-source"] for row in rows) + assert "log" in sources + assert "issue" in sources + + def test_run_with_config(self, log_csv, tmp_path): + output = tmp_path / "tasks.csv" + config = TaskPipelineConfig( + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + output_path=output, + log_path=log_csv, + ) + status = TaskPipeline(config).run() + assert status == TaskPipelineStatus.SUCCESS + assert len(_read_csv(output)) == 2 + + def test_run_output_has_correct_columns(self, log_csv, tmp_path): + output = tmp_path / "tasks.csv" + TaskPipeline().run( + output_path=output, + dataset="tree-preservation-zone", + organisation="local-authority-eng:ABC", + endpoint="endpoint-aaa", + log_path=log_csv, + ) + expected_cols = { + "dataset", + "organisation", + "endpoint", + "resource", + "details", + "severity", + "responsibility", + "task-source", + "entry-date", + "task-id", + } + rows = _read_csv(output) + assert set(rows[0].keys()) == expected_cols diff --git a/tests/unit/test_task_pipeline.py b/tests/unit/test_task_pipeline.py deleted file mode 100644 index 079430e72..000000000 --- a/tests/unit/test_task_pipeline.py +++ /dev/null @@ -1,183 +0,0 @@ -import csv -import json -import pytest - -from digital_land.pipeline.task import ( - TaskPipeline, - TaskPipelineConfig, - TaskPipelineStatus, -) - - -@pytest.fixture -def log_csv(tmp_path): - """A collection log with one success and two failures.""" - path = tmp_path / "log.csv" - path.write_text( - "endpoint,resource,status,exception,entry-date,bytes,elapsed\n" - "endpoint-aaa,resource-aaa,200,,2024-01-01,1234,0.5\n" - "endpoint-bbb,resource-bbb,404,,2024-01-01,0,0.1\n" - "endpoint-ccc,,500,Connection refused,2024-01-01,0,0.0\n" - ) - return path - - -@pytest.fixture -def issue_csv(tmp_path): - """An issue log with error/external issues and one warning to be filtered out.""" - path = tmp_path / "issue.csv" - path.write_text( - "dataset,resource,line-number,entry-number,field,entity,issue-type,value,message,severity,responsibility\n" - "tree-preservation-zone,resource-aaa,1,1,geometry,,invalid-geometry,bad-wkt,Invalid geometry,error,external\n" - "tree-preservation-zone,resource-aaa,2,2,geometry,,invalid-geometry,other-bad-wkt,Invalid geometry,error,external\n" - "tree-preservation-zone,resource-aaa,3,3,name,,missing-value,,Missing value,error,external\n" - "tree-preservation-zone,resource-aaa,4,4,notes,,missing-value,,Missing value,warning,internal\n" - ) - return path - - -def _read_csv(path): - with open(path) as f: - return list(csv.DictReader(f)) - - -def test_log_tasks_creates_row_per_failure(log_csv, tmp_path): - output = tmp_path / "tasks.csv" - status = TaskPipeline().run( - output_path=output, - dataset="tree-preservation-zone", - organisation="local-authority-eng:ABC", - endpoint="endpoint-aaa", - log_path=log_csv, - ) - - assert status == TaskPipelineStatus.SUCCESS - rows = _read_csv(output) - assert len(rows) == 2 - assert set(row["task-source"] for row in rows) == {"log"} - - -def test_log_task_details_json(log_csv, tmp_path): - output = tmp_path / "tasks.csv" - TaskPipeline().run( - output_path=output, - dataset="tree-preservation-zone", - organisation="local-authority-eng:ABC", - endpoint="endpoint-aaa", - log_path=log_csv, - ) - - rows = _read_csv(output) - details = [json.loads(row["details"]) for row in rows] - statuses = {detail["status"] for detail in details} - assert 404 in statuses - assert 500 in statuses - assert all(isinstance(d["status"], int) for d in details) - - -def test_log_tasks_empty_when_all_successful(tmp_path): - log = tmp_path / "log.csv" - log.write_text( - "endpoint,resource,status,exception,entry-date,bytes,elapsed\n" - "endpoint-aaa,resource-aaa,200,,2024-01-01,1234,0.5\n" - ) - output = tmp_path / "tasks.csv" - status = TaskPipeline().run( - output_path=output, - dataset="tree-preservation-zone", - organisation="local-authority-eng:ABC", - endpoint="endpoint-aaa", - log_path=log, - ) - assert status == TaskPipelineStatus.NO_TASKS - assert _read_csv(output) == [] - - -def test_issue_tasks_groups_by_type_and_field(issue_csv, tmp_path): - output = tmp_path / "tasks.csv" - status = TaskPipeline().run( - output_path=output, - dataset="tree-preservation-zone", - organisation="local-authority-eng:ABC", - endpoint="endpoint-aaa", - issue_path=issue_csv, - ) - - assert status == TaskPipelineStatus.SUCCESS - rows = _read_csv(output) - assert len(rows) == 2 - assert set(row["task-source"] for row in rows) == {"issue"} - - -def test_issue_task_details_json(issue_csv, tmp_path): - output = tmp_path / "tasks.csv" - TaskPipeline().run( - output_path=output, - dataset="tree-preservation-zone", - organisation="local-authority-eng:ABC", - endpoint="endpoint-aaa", - issue_path=issue_csv, - ) - - rows = _read_csv(output) - details = [json.loads(row["details"]) for row in rows] - geom_task = next(d for d in details if d["issue_type"] == "invalid-geometry") - assert geom_task["count"] == 2 - assert geom_task["field"] == "geometry" - - -def test_both_sources_combined(log_csv, issue_csv, tmp_path): - output = tmp_path / "tasks.csv" - status = TaskPipeline().run( - output_path=output, - dataset="tree-preservation-zone", - organisation="local-authority-eng:ABC", - endpoint="endpoint-aaa", - log_path=log_csv, - issue_path=issue_csv, - ) - - assert status == TaskPipelineStatus.SUCCESS - rows = _read_csv(output) - sources = set(row["task-source"] for row in rows) - assert "log" in sources - assert "issue" in sources - - -def test_run_with_config(log_csv, tmp_path): - output = tmp_path / "tasks.csv" - config = TaskPipelineConfig( - dataset="tree-preservation-zone", - organisation="local-authority-eng:ABC", - endpoint="endpoint-aaa", - output_path=output, - log_path=log_csv, - ) - status = TaskPipeline(config).run() - assert status == TaskPipelineStatus.SUCCESS - assert len(_read_csv(output)) == 2 - - -def test_output_has_correct_columns(log_csv, tmp_path): - output = tmp_path / "tasks.csv" - TaskPipeline().run( - output_path=output, - dataset="tree-preservation-zone", - organisation="local-authority-eng:ABC", - endpoint="endpoint-aaa", - log_path=log_csv, - ) - expected_cols = { - "dataset", - "organisation", - "endpoint", - "resource", - "details", - "severity", - "responsibility", - "task-source", - "entry-date", - "task-id", - } - rows = _read_csv(output) - assert set(rows[0].keys()) == expected_cols From 3c730efda0c2e6933327f9c3553157564aeec095 Mon Sep 17 00:00:00 2001 From: Tom Brooks Date: Fri, 15 May 2026 14:13:24 +0100 Subject: [PATCH 5/7] refactor: move I/O logic to run from helper functions --- digital_land/pipeline/task.py | 54 ++++++++++++++++++----------------- 1 file changed, 28 insertions(+), 26 deletions(-) diff --git a/digital_land/pipeline/task.py b/digital_land/pipeline/task.py index 382e0f2a9..44995d023 100644 --- a/digital_land/pipeline/task.py +++ b/digital_land/pipeline/task.py @@ -90,13 +90,25 @@ def run( frames = [] if log_path and Path(log_path).exists(): - log_tasks = _tasks_from_log(log_path, dataset, organisation, entry_date) + log_df = ( + # infer_schema_length=0 means that polas will interpret all values as strings + # rather than the defaul of infering datatypes on first 100 rows + pl.scan_csv( + log_path, infer_schema_length=0, null_values=[""] + ).collect() + ) + log_tasks = _transform_log_to_tasks( + log_df, dataset, organisation, entry_date + ) if not log_tasks.is_empty(): frames.append(log_tasks) if issue_path and Path(issue_path).exists(): - issue_tasks = _tasks_from_issues( - issue_path, + issue_df = pl.scan_csv( + issue_path, infer_schema_length=0, null_values=[""] + ).collect() + issue_tasks = _transform_issues_to_tasks( + issue_df, organisation, endpoint, severity_filter, @@ -120,22 +132,15 @@ def run( return TaskPipelineStatus.FAILED -def _tasks_from_log( - log_path: Path, +def _transform_log_to_tasks( + df: pl.DataFrame, dataset: str, organisation: str, entry_date: str, ) -> pl.DataFrame: - """Return a task row for each failed collection log entry.""" - - # scan_csv + filter stays lazy; .collect() materialises only the failed rows - # infer_schema_length=0 tells polars to read all columns as strings rather than - # infer datatypes based on the first 100 rows, the default behaviour. - failed = ( - pl.scan_csv(log_path, infer_schema_length=0, null_values=[""]) - .filter(pl.col("status") != "200") - .collect() - ) + """Filter and transform a log DataFrame into task rows.""" + + failed = df.filter(pl.col("status") != "200") if failed.is_empty(): return pl.DataFrame(schema=_EMPTY_SCHEMA) @@ -175,7 +180,7 @@ def _tasks_from_log( "entry-date": pl.Series([entry_date] * n, dtype=pl.Utf8), } ) - # callculate a sha256 hash from some key columns to identify this task + # calculate a sha256 hash from some key columns to identify this task result = result.with_columns( pl.struct(["dataset", "endpoint", "resource", "task-source", "details"]) .map_elements( @@ -197,25 +202,22 @@ def _tasks_from_log( return result -def _tasks_from_issues( - issue_path: Path, +def _transform_issues_to_tasks( + df: pl.DataFrame, organisation: str, endpoint: str, severity_filter: List[str], responsibility_filter: List[str], entry_date: str, ) -> pl.DataFrame: - """Return one task row per (issue-type, resource, field, dataset) group.""" + """Filter and transform an issue DataFrame into task rows.""" - lf = pl.scan_csv(issue_path, infer_schema_length=0, null_values=[""]) - cols = set(lf.columns) # .columns works on LazyFrame without collecting + cols = set(df.columns) if "severity" in cols: - lf = lf.filter(pl.col("severity").is_in(severity_filter)) + df = df.filter(pl.col("severity").is_in(severity_filter)) if "responsibility" in cols: - lf = lf.filter(pl.col("responsibility").is_in(responsibility_filter)) - - df = lf.collect() # materialise after filters are applied + df = df.filter(pl.col("responsibility").is_in(responsibility_filter)) if df.is_empty(): return pl.DataFrame(schema=_EMPTY_SCHEMA) @@ -271,7 +273,7 @@ def _tasks_from_issues( pl.lit(entry_date).alias("entry-date"), ] ) - # callculate a sha256 hash from some key columns to identify this task + # calculate a sha256 hash from some key columns to identify this task result = result.with_columns( pl.struct(["dataset", "endpoint", "resource", "task-source", "details"]) .map_elements( From abfa511915ec58b7be633351c7ef4e9caf5e862a Mon Sep 17 00:00:00 2001 From: Tom Brooks Date: Fri, 15 May 2026 14:24:53 +0100 Subject: [PATCH 6/7] Change task-id to reference --- digital_land/pipeline/task.py | 6 +++--- tests/unit/pipeline/test_task.py | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/digital_land/pipeline/task.py b/digital_land/pipeline/task.py index 44995d023..f13984519 100644 --- a/digital_land/pipeline/task.py +++ b/digital_land/pipeline/task.py @@ -24,7 +24,7 @@ "responsibility", "task-source", "entry-date", - "task-id", + "reference", ] _EMPTY_SCHEMA = {col: pl.Utf8 for col in TASK_COLUMNS} @@ -197,7 +197,7 @@ def _transform_log_to_tasks( ).hexdigest()[:16], return_dtype=pl.Utf8, ) - .alias("task-id") + .alias("reference") ) return result @@ -290,6 +290,6 @@ def _transform_issues_to_tasks( ).hexdigest()[:16], return_dtype=pl.Utf8, ) - .alias("task-id") + .alias("reference") ) return result diff --git a/tests/unit/pipeline/test_task.py b/tests/unit/pipeline/test_task.py index a969b74ad..48d5648e9 100644 --- a/tests/unit/pipeline/test_task.py +++ b/tests/unit/pipeline/test_task.py @@ -172,7 +172,7 @@ def test_run_output_has_correct_columns(self, log_csv, tmp_path): "responsibility", "task-source", "entry-date", - "task-id", + "reference", } rows = _read_csv(output) assert set(rows[0].keys()) == expected_cols From 29cfe267600340eff1db67b303f8d3f3d1ff98ba Mon Sep 17 00:00:00 2001 From: Tom Brooks Date: Fri, 15 May 2026 16:07:03 +0100 Subject: [PATCH 7/7] make TaskPipelineStatus more similar to PipelineStatus --- digital_land/pipeline/task.py | 13 ++++++------- tests/unit/pipeline/test_task.py | 10 +++++----- 2 files changed, 11 insertions(+), 12 deletions(-) diff --git a/digital_land/pipeline/task.py b/digital_land/pipeline/task.py index f13984519..67b8f13be 100644 --- a/digital_land/pipeline/task.py +++ b/digital_land/pipeline/task.py @@ -31,8 +31,8 @@ class TaskPipelineStatus(Enum): - SUCCESS = 1 - NO_TASKS = 2 + INITIALISED = 1 + COMPLETE = 2 FAILED = 3 @@ -54,6 +54,7 @@ class TaskPipeline: def __init__(self, config: Optional[TaskPipelineConfig] = None): self.config = config + self.status = TaskPipelineStatus.INITIALISED def run( self, @@ -121,14 +122,12 @@ def run( result = pl.concat(frames) if frames else pl.DataFrame(schema=_EMPTY_SCHEMA) result.write_csv(output_path) - return ( - TaskPipelineStatus.NO_TASKS - if result.is_empty() - else TaskPipelineStatus.SUCCESS - ) + self.status = TaskPipelineStatus.COMPLETE + return TaskPipelineStatus.COMPLETE except Exception: logger.exception("TaskPipeline failed") + self.status = TaskPipelineStatus.FAILED return TaskPipelineStatus.FAILED diff --git a/tests/unit/pipeline/test_task.py b/tests/unit/pipeline/test_task.py index 48d5648e9..9e9ef50c8 100644 --- a/tests/unit/pipeline/test_task.py +++ b/tests/unit/pipeline/test_task.py @@ -53,7 +53,7 @@ def test_run_log_tasks_creates_row_per_failure(self, log_csv, tmp_path): log_path=log_csv, ) - assert status == TaskPipelineStatus.SUCCESS + assert status == TaskPipelineStatus.COMPLETE rows = _read_csv(output) assert len(rows) == 2 assert set(row["task-source"] for row in rows) == {"log"} @@ -89,7 +89,7 @@ def test_run_log_tasks_empty_when_all_successful(self, tmp_path): endpoint="endpoint-aaa", log_path=log, ) - assert status == TaskPipelineStatus.NO_TASKS + assert status == TaskPipelineStatus.COMPLETE assert _read_csv(output) == [] def test_run_issue_tasks_groups_by_type_and_field(self, issue_csv, tmp_path): @@ -102,7 +102,7 @@ def test_run_issue_tasks_groups_by_type_and_field(self, issue_csv, tmp_path): issue_path=issue_csv, ) - assert status == TaskPipelineStatus.SUCCESS + assert status == TaskPipelineStatus.COMPLETE rows = _read_csv(output) assert len(rows) == 2 assert set(row["task-source"] for row in rows) == {"issue"} @@ -134,7 +134,7 @@ def test_run_both_sources_combined(self, log_csv, issue_csv, tmp_path): issue_path=issue_csv, ) - assert status == TaskPipelineStatus.SUCCESS + assert status == TaskPipelineStatus.COMPLETE rows = _read_csv(output) sources = set(row["task-source"] for row in rows) assert "log" in sources @@ -150,7 +150,7 @@ def test_run_with_config(self, log_csv, tmp_path): log_path=log_csv, ) status = TaskPipeline(config).run() - assert status == TaskPipelineStatus.SUCCESS + assert status == TaskPipelineStatus.COMPLETE assert len(_read_csv(output)) == 2 def test_run_output_has_correct_columns(self, log_csv, tmp_path):