Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
550 lines
21 KiB
Python
550 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from geoalchemy2.shape import from_shape, to_shape
|
|
from shapely.geometry import Polygon, box
|
|
|
|
from app.core.errors import AppError
|
|
from app.models import Dataset, DatasetVersion
|
|
from app.schemas.dataset import DatasetTemporalUpdate
|
|
from app.schemas.temporal import TemporalComparisonRequest, TemporalObjectChanges
|
|
from app.services.dataset_service import DatasetService
|
|
from app.services.temporal_analysis_service import TemporalAnalysisService
|
|
from app.services.vector_feature_service import VectorFeatureService
|
|
from tests.frontend_contract import read_map_workspace
|
|
|
|
|
|
ROOT = Path(__file__).parents[2]
|
|
|
|
|
|
class ScalarQuery:
|
|
def __init__(self, value: float):
|
|
self.value = value
|
|
|
|
def filter(self, *args): # noqa: ANN002, ARG002
|
|
return self
|
|
|
|
def scalar(self):
|
|
return self.value
|
|
|
|
|
|
class ScalarSession:
|
|
def __init__(self, value: float):
|
|
self.value = value
|
|
|
|
def query(self, *args): # noqa: ANN002, ARG002
|
|
return ScalarQuery(self.value)
|
|
|
|
|
|
class SequenceScalarSession:
|
|
def __init__(self, values: list[float]):
|
|
self.values = iter(values)
|
|
|
|
def query(self, *args): # noqa: ANN002, ARG002
|
|
return ScalarQuery(next(self.values))
|
|
|
|
|
|
class FeatureRowsQuery:
|
|
def __init__(self, rows: list[object]):
|
|
self.rows = rows
|
|
self.row_limit: int | None = None
|
|
|
|
def filter(self, *args): # noqa: ANN002, ARG002
|
|
return self
|
|
|
|
def order_by(self, *args): # noqa: ANN002, ARG002
|
|
return self
|
|
|
|
def limit(self, value: int):
|
|
self.row_limit = value
|
|
return self
|
|
|
|
def all(self):
|
|
return self.rows[: self.row_limit]
|
|
|
|
|
|
class SequentialFeatureSession:
|
|
def __init__(self, row_sets: list[list[object]]):
|
|
self.row_sets = iter(row_sets)
|
|
|
|
def query(self, _model):
|
|
return FeatureRowsQuery(next(self.row_sets))
|
|
|
|
|
|
class VersionQuery:
|
|
def __init__(self, latest: DatasetVersion | None):
|
|
self.latest = latest
|
|
|
|
def filter(self, *args): # noqa: ANN002, ARG002
|
|
return self
|
|
|
|
def order_by(self, *args): # noqa: ANN002, ARG002
|
|
return self
|
|
|
|
def first(self):
|
|
return self.latest
|
|
|
|
|
|
class TemporalUpdateSession:
|
|
def __init__(self, dataset: Dataset, latest: DatasetVersion | None):
|
|
self.dataset = dataset
|
|
self.latest = latest
|
|
self.added: list[object] = []
|
|
|
|
def get(self, model, item_id): # noqa: ANN001
|
|
return self.dataset if model is Dataset and item_id == self.dataset.id else None
|
|
|
|
def query(self, model): # noqa: ANN001
|
|
assert model is DatasetVersion
|
|
return VersionQuery(self.latest)
|
|
|
|
def add(self, item): # noqa: ANN001
|
|
self.added.append(item)
|
|
|
|
def commit(self):
|
|
return None
|
|
|
|
def refresh(self, _item):
|
|
return None
|
|
|
|
|
|
def temporal_dataset(*, project_id, observed_year: int, metric_method: str = "feature_count") -> Dataset:
|
|
return Dataset(
|
|
id=uuid4(),
|
|
project_id=project_id,
|
|
name=f"snapshot-{observed_year}.geojson",
|
|
dataset_type="vector",
|
|
source="official",
|
|
dataset_role="reference",
|
|
temporal_series_key="official:test:mol",
|
|
observed_at=datetime(observed_year, 1, 1, tzinfo=timezone.utc),
|
|
source_version=str(observed_year),
|
|
source_metadata={
|
|
"selection_aggregation": {
|
|
"method": metric_method,
|
|
"label": "Objecten",
|
|
"unit": "objecten",
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
def test_temporal_series_keeps_only_latest_snapshot_per_observation_date() -> None:
|
|
project_id = uuid4()
|
|
old = temporal_dataset(project_id=project_id, observed_year=2025)
|
|
old.imported_at = datetime(2026, 7, 19, tzinfo=timezone.utc)
|
|
latest = temporal_dataset(project_id=project_id, observed_year=2025)
|
|
latest.imported_at = datetime(2026, 7, 21, tzinfo=timezone.utc)
|
|
earlier = temporal_dataset(project_id=project_id, observed_year=2022)
|
|
earlier.imported_at = datetime(2026, 7, 21, tzinfo=timezone.utc)
|
|
|
|
canonical = TemporalAnalysisService._canonical_observation_snapshots([old, latest, earlier])
|
|
|
|
assert [dataset.id for dataset in canonical] == [earlier.id, latest.id]
|
|
|
|
|
|
def governed_grb_dataset(*, project_id, observed_day: int) -> Dataset:
|
|
dataset = temporal_dataset(project_id=project_id, observed_year=2026, metric_method="intersection_area")
|
|
dataset.observed_at = datetime(2026, 7, observed_day, tzinfo=timezone.utc)
|
|
dataset.source_version = f"2026-07-{observed_day:02d}"
|
|
dataset.source_name = "grb"
|
|
dataset.reference_layer_name = "buildings"
|
|
dataset.temporal_series_key = "grb:buildings:kempen-transport-region"
|
|
dataset.source_metadata = {
|
|
"authority_level": "authoritative",
|
|
"collection": "GRB/GBG",
|
|
"coverage_scope": "kempen-transport-region",
|
|
"scope_type": "transport_region",
|
|
"member_count": 28,
|
|
"partition_count": 28,
|
|
"partition_strategy": "municipality_bbox_maximum_boundary_intersection",
|
|
"selection_aggregation": {
|
|
"method": "intersection_area",
|
|
"label": "Bebouwde grondoppervlakte",
|
|
"unit": "ha",
|
|
},
|
|
}
|
|
dataset.provenance_metadata = {
|
|
"operator_tool": "provision_regional_grb_buildings.py",
|
|
"reference_truncated": False,
|
|
"manifest_path": "/storage/operator/grb/manifest.json",
|
|
"source_url": "https://geo.api.vlaanderen.be/GRB/ogc/features/v1/collections/GBG/items",
|
|
"artifact_sha256": "a" * 64,
|
|
"partition_checksums": {f"{index:05d}": "b" * 64 for index in range(28)},
|
|
}
|
|
if observed_day > 14:
|
|
dataset.source_metadata["geometry_clipped_to_area"] = True
|
|
return dataset
|
|
|
|
|
|
def persisted_feature(dataset_id, source_feature_id: str | None, polygon: Polygon):
|
|
return SimpleNamespace(
|
|
id=uuid4(),
|
|
dataset_id=dataset_id,
|
|
source_feature_id=source_feature_id,
|
|
properties_json={},
|
|
geometry=from_shape(polygon, srid=4326),
|
|
)
|
|
|
|
|
|
def test_temporal_migration_and_models_align() -> None:
|
|
migration = (ROOT / "backend/alembic/versions/202607140001_temporal_dataset_foundation.py").read_text(encoding="utf-8")
|
|
for field in (
|
|
"temporal_series_key",
|
|
"observed_at",
|
|
"valid_from",
|
|
"valid_to",
|
|
"temporal_granularity",
|
|
"source_version",
|
|
):
|
|
assert field in migration
|
|
assert hasattr(Dataset, field)
|
|
assert "ix_vector_features_dataset_source_feature" in migration
|
|
assert 'down_revision = "202606120900"' in migration
|
|
|
|
|
|
def test_temporal_metadata_requires_an_explicit_series_and_observation_date() -> None:
|
|
with pytest.raises(AppError, match="observed_at is required"):
|
|
DatasetService._validate_temporal_metadata(
|
|
temporal_series_key="official:test:mol",
|
|
observed_at=None,
|
|
valid_from=None,
|
|
valid_to=None,
|
|
temporal_granularity="year",
|
|
source_version="2024",
|
|
)
|
|
with pytest.raises(AppError, match="valid_to must be"):
|
|
DatasetService._validate_temporal_metadata(
|
|
temporal_series_key="official:test:mol",
|
|
observed_at=datetime(2024, 1, 1, tzinfo=timezone.utc),
|
|
valid_from=datetime(2024, 12, 31, tzinfo=timezone.utc),
|
|
valid_to=datetime(2024, 1, 1, tzinfo=timezone.utc),
|
|
temporal_granularity="year",
|
|
source_version="2024",
|
|
)
|
|
|
|
|
|
def test_temporal_metadata_update_appends_provenance_version_and_is_idempotent() -> None:
|
|
project_id = uuid4()
|
|
dataset = temporal_dataset(project_id=project_id, observed_year=2024)
|
|
dataset.status = "ready"
|
|
dataset.metadata_json = {}
|
|
latest = DatasetVersion(
|
|
dataset_id=dataset.id,
|
|
version=3,
|
|
observed_at=dataset.observed_at,
|
|
source_version="2024",
|
|
)
|
|
session = TemporalUpdateSession(dataset, latest)
|
|
payload = DatasetTemporalUpdate(
|
|
temporal_series_key="official:test:mol",
|
|
observed_at=datetime(2025, 1, 1, tzinfo=timezone.utc),
|
|
temporal_granularity="year",
|
|
source_version="2025",
|
|
)
|
|
|
|
updated = DatasetService.update_temporal_metadata(session, dataset.id, payload)
|
|
|
|
assert updated.observed_at == payload.observed_at
|
|
assert latest.version == 3
|
|
assert latest.observed_at == datetime(2024, 1, 1, tzinfo=timezone.utc)
|
|
assert len(session.added) == 2
|
|
appended = session.added[1]
|
|
assert isinstance(appended, DatasetVersion)
|
|
assert appended.version == 4
|
|
assert appended.observed_at == payload.observed_at
|
|
|
|
session.added.clear()
|
|
DatasetService.update_temporal_metadata(session, dataset.id, payload)
|
|
assert session.added == []
|
|
|
|
|
|
def test_selection_area_aggregation_returns_hectares_without_loading_all_features() -> None:
|
|
project_id = uuid4()
|
|
dataset = temporal_dataset(project_id=project_id, observed_year=1969, metric_method="intersection_area")
|
|
dataset.source_metadata["selection_aggregation"].update({"label": "Oppervlakte", "unit": "ha"})
|
|
result = VectorFeatureService.summarize_features_by_bbox(
|
|
ScalarSession(125_000.0),
|
|
dataset=dataset,
|
|
bbox={"min_x": 5.0, "min_y": 51.1, "max_x": 5.2, "max_y": 51.3, "crs": "EPSG:4326"},
|
|
total_feature_count=40,
|
|
)
|
|
assert result["metric_value"] == 12.5
|
|
assert result["metric_unit"] == "ha"
|
|
assert result["feature_count"] == 40
|
|
|
|
|
|
def test_population_area_weighting_is_exact_for_full_features_and_estimated_for_partial_features() -> None:
|
|
dataset = temporal_dataset(project_id=uuid4(), observed_year=2025, metric_method="area_weighted_sum")
|
|
dataset.source_metadata["selection_aggregation"].update(
|
|
{
|
|
"property": "population_total",
|
|
"label": "Inwoners",
|
|
"unit": "inwoners",
|
|
"warning": "Partial-sector estimate",
|
|
"warning_only_when_estimate": True,
|
|
}
|
|
)
|
|
bbox = {"min_x": 5.0, "min_y": 51.1, "max_x": 5.2, "max_y": 51.3, "crs": "EPSG:4326"}
|
|
|
|
full = VectorFeatureService.summarize_features_by_bbox(
|
|
SequenceScalarSession([49, 38_675.0]),
|
|
dataset=dataset,
|
|
bbox=bbox,
|
|
total_feature_count=49,
|
|
)
|
|
partial = VectorFeatureService.summarize_features_by_bbox(
|
|
SequenceScalarSession([1, 1_250.5]),
|
|
dataset=dataset,
|
|
bbox=bbox,
|
|
total_feature_count=3,
|
|
)
|
|
|
|
assert full["metric_value"] == 38_675.0
|
|
assert full["is_estimate"] is False
|
|
assert full["warning"] is None
|
|
assert partial["metric_value"] == 1_250.5
|
|
assert partial["is_estimate"] is True
|
|
assert partial["warning"] == "Partial-sector estimate"
|
|
|
|
|
|
def test_temporal_compare_returns_delta_and_canonical_change_payload(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
earlier = temporal_dataset(project_id=project_id, observed_year=2021)
|
|
later = temporal_dataset(project_id=project_id, observed_year=2024)
|
|
|
|
def get_dataset(_db, _project_id, dataset_id, _label):
|
|
return earlier if dataset_id == earlier.id else later
|
|
|
|
def summarize(_db, *, dataset, bbox, disclose_selection_edge=True): # noqa: ARG001
|
|
value = 100.0 if dataset.id == earlier.id else 115.0
|
|
return {
|
|
"metric_label": "Inwoners",
|
|
"metric_value": value,
|
|
"metric_unit": "inwoners",
|
|
"aggregation_method": "area_weighted_sum",
|
|
"feature_count": 10,
|
|
"is_estimate": True,
|
|
"warning": "Areal weighting",
|
|
}
|
|
|
|
monkeypatch.setattr(TemporalAnalysisService, "_get_temporal_dataset", staticmethod(get_dataset))
|
|
monkeypatch.setattr(VectorFeatureService, "summarize_features_by_bbox", staticmethod(summarize))
|
|
monkeypatch.setattr(
|
|
TemporalAnalysisService,
|
|
"_compare_identity_features",
|
|
staticmethod(
|
|
lambda *args, **kwargs: (
|
|
TemporalObjectChanges(available=True, added_count=1, removed_count=0, modified_count=2, unchanged_count=7),
|
|
{"type": "FeatureCollection", "features": []},
|
|
[],
|
|
)
|
|
),
|
|
)
|
|
result = TemporalAnalysisService.compare(
|
|
SimpleNamespace(),
|
|
project_id=project_id,
|
|
payload=TemporalComparisonRequest(
|
|
earlier_dataset_id=earlier.id,
|
|
later_dataset_id=later.id,
|
|
bbox={"min_x": 5.0, "min_y": 51.1, "max_x": 5.2, "max_y": 51.3},
|
|
),
|
|
)
|
|
assert result.metric.absolute_change == 15.0
|
|
assert result.metric.percent_change == 15.0
|
|
assert result.metric.is_estimate is True
|
|
assert result.object_changes.modified_count == 2
|
|
assert result.geojson["type"] == "FeatureCollection"
|
|
|
|
|
|
def test_temporal_comparison_clips_cross_boundary_bbox_to_selected_area(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
area_id = uuid4()
|
|
earlier = temporal_dataset(project_id=project_id, observed_year=2021)
|
|
later = temporal_dataset(project_id=project_id, observed_year=2024)
|
|
area_shape = box(5.0, 51.0, 5.2, 51.2)
|
|
area = SimpleNamespace(
|
|
id=area_id,
|
|
project_id=project_id,
|
|
geometry=from_shape(area_shape, srid=4326),
|
|
)
|
|
captured_geometries = []
|
|
identity_capture = {}
|
|
|
|
monkeypatch.setattr(
|
|
TemporalAnalysisService,
|
|
"_get_temporal_dataset",
|
|
staticmethod(lambda _db, _project_id, dataset_id, _label: earlier if dataset_id == earlier.id else later),
|
|
)
|
|
monkeypatch.setattr(
|
|
TemporalAnalysisService,
|
|
"_get_selection_area",
|
|
staticmethod(lambda _db, _project_id, requested_area_id: area if requested_area_id == area_id else None),
|
|
)
|
|
|
|
def summarize(_db, *, dataset, bbox, selection_geometry, full_dataset_area, disclose_selection_edge=True): # noqa: ARG001
|
|
captured_geometries.append(selection_geometry)
|
|
return {
|
|
"metric_label": "Oppervlakte",
|
|
"metric_value": 10.0 if dataset.id == earlier.id else 12.0,
|
|
"metric_unit": "ha",
|
|
"aggregation_method": "intersection_area",
|
|
"feature_count": 1,
|
|
"is_estimate": False,
|
|
"warning": None,
|
|
}
|
|
|
|
def compare_identity(*_args, **kwargs):
|
|
identity_capture.update(kwargs)
|
|
return (
|
|
TemporalObjectChanges(available=False),
|
|
{"type": "FeatureCollection", "features": []},
|
|
[],
|
|
)
|
|
|
|
monkeypatch.setattr(VectorFeatureService, "summarize_features_by_bbox", staticmethod(summarize))
|
|
monkeypatch.setattr(TemporalAnalysisService, "_compare_identity_features", staticmethod(compare_identity))
|
|
|
|
result = TemporalAnalysisService.compare(
|
|
SimpleNamespace(),
|
|
project_id=project_id,
|
|
payload=TemporalComparisonRequest(
|
|
earlier_dataset_id=earlier.id,
|
|
later_dataset_id=later.id,
|
|
area_id=area_id,
|
|
bbox={"min_x": 4.9, "min_y": 51.1, "max_x": 5.1, "max_y": 51.3},
|
|
),
|
|
)
|
|
|
|
expected = box(5.0, 51.1, 5.1, 51.2)
|
|
assert result.metric.absolute_change == 2.0
|
|
assert all(to_shape(geometry).equals(expected) for geometry in captured_geometries)
|
|
assert to_shape(identity_capture["selection_geometry"]).equals(expected)
|
|
assert identity_capture["earlier_full_dataset_area"] is False
|
|
assert identity_capture["later_full_dataset_area"] is False
|
|
|
|
|
|
def test_unstable_temporal_identity_returns_clear_end_user_warning() -> None:
|
|
project_id = uuid4()
|
|
earlier = temporal_dataset(project_id=project_id, observed_year=2021)
|
|
later = temporal_dataset(project_id=project_id, observed_year=2025)
|
|
earlier.source_metadata["identity_stable"] = False
|
|
later.source_metadata["identity_stable"] = False
|
|
|
|
changes, geojson, warnings = TemporalAnalysisService._compare_identity_features(
|
|
SimpleNamespace(),
|
|
earlier=earlier,
|
|
later=later,
|
|
bbox={"min_x": 5.0, "min_y": 51.1, "max_x": 5.2, "max_y": 51.3},
|
|
preview_limit=100,
|
|
)
|
|
|
|
assert changes.available is False
|
|
assert geojson == {"type": "FeatureCollection", "features": []}
|
|
assert warnings == ["Wijzigingen van individuele objecten kunnen voor deze bron niet betrouwbaar worden gevolgd."]
|
|
|
|
|
|
def test_governed_legacy_grb_snapshots_use_verified_official_feature_identity() -> None:
|
|
project_id = uuid4()
|
|
earlier = governed_grb_dataset(project_id=project_id, observed_day=14)
|
|
later = governed_grb_dataset(project_id=project_id, observed_day=15)
|
|
original = Polygon([(5.1, 51.1), (5.101, 51.1), (5.101, 51.101), (5.1, 51.101)])
|
|
changed = Polygon([(5.1, 51.1), (5.102, 51.1), (5.102, 51.101), (5.1, 51.101)])
|
|
added = Polygon([(5.11, 51.11), (5.111, 51.11), (5.111, 51.111), (5.11, 51.111)])
|
|
session = SequentialFeatureSession(
|
|
[
|
|
[persisted_feature(earlier.id, "GBG.1", original)],
|
|
[persisted_feature(later.id, "GBG.1", changed), persisted_feature(later.id, "GBG.2", added)],
|
|
]
|
|
)
|
|
|
|
changes, geojson, warnings = TemporalAnalysisService._compare_identity_features(
|
|
session,
|
|
earlier=earlier,
|
|
later=later,
|
|
bbox={"min_x": 5.0, "min_y": 51.0, "max_x": 5.2, "max_y": 51.2},
|
|
preview_limit=100,
|
|
)
|
|
|
|
assert changes.available is True
|
|
assert changes.added_count == 1
|
|
assert changes.removed_count == 0
|
|
assert changes.modified_count == 1
|
|
assert changes.unchanged_count == 0
|
|
assert {feature["properties"]["change_type"] for feature in geojson["features"]} == {"added", "modified"}
|
|
assert warnings == []
|
|
|
|
|
|
def test_governed_grb_object_history_fails_closed_for_unverified_identity() -> None:
|
|
project_id = uuid4()
|
|
earlier = governed_grb_dataset(project_id=project_id, observed_day=14)
|
|
later = governed_grb_dataset(project_id=project_id, observed_day=15)
|
|
polygon = Polygon([(5.1, 51.1), (5.101, 51.1), (5.101, 51.101), (5.1, 51.101)])
|
|
fallback_hash = "a" * 64
|
|
session = SequentialFeatureSession(
|
|
[
|
|
[persisted_feature(earlier.id, fallback_hash, polygon)],
|
|
[persisted_feature(later.id, fallback_hash, polygon)],
|
|
]
|
|
)
|
|
|
|
changes, geojson, warnings = TemporalAnalysisService._compare_identity_features(
|
|
session,
|
|
earlier=earlier,
|
|
later=later,
|
|
bbox={"min_x": 5.0, "min_y": 51.0, "max_x": 5.2, "max_y": 51.2},
|
|
preview_limit=100,
|
|
)
|
|
|
|
assert changes.available is False
|
|
assert geojson["features"] == []
|
|
assert warnings == ["De geselecteerde objecten bevatten geen volledig verifieerbare stabiele bronidentiteit."]
|
|
|
|
|
|
def test_legacy_grb_identity_requires_complete_partition_evidence() -> None:
|
|
dataset = governed_grb_dataset(project_id=uuid4(), observed_day=14)
|
|
dataset.provenance_metadata["partition_checksums"] = {"13025": "b" * 64}
|
|
|
|
assert TemporalAnalysisService._identity_contract(dataset) is None
|
|
|
|
|
|
def test_temporal_frontend_and_official_operator_contracts_exist() -> None:
|
|
workspace = read_map_workspace()
|
|
temporal_api = (ROOT / "frontend/src/services/api/temporal.ts").read_text(encoding="utf-8")
|
|
population = (ROOT / "scripts/provision_mol_population_history.py").read_text(encoding="utf-8")
|
|
landuse = (ROOT / "scripts/provision_mol_historical_landuse.py").read_text(encoding="utf-8")
|
|
dockerfile = (ROOT / "deploy/unraid/Dockerfile.all-in-one").read_text(encoding="utf-8")
|
|
|
|
assert "Laatste toestand" in workspace
|
|
assert "Evolutie" in workspace
|
|
assert "Vergelijk periode" in workspace
|
|
assert "temporalRangeLabel" in workspace
|
|
assert "Dagelijkse GRB-edities tonen wijzigingen in de officiële registratie" in workspace
|
|
assert "/temporal/compare" in temporal_api
|
|
assert "Statbel" in population and "area_weighted_sum" in population
|
|
assert '"identity_stable": False' in population
|
|
assert "HistLandgebruik" in landuse and "intersection_area" in landuse
|
|
assert "<wfs:GetFeature" in landuse and "session.post(" in landuse
|
|
assert 'lambda value: value.startswith("weg"), "weg*"' in landuse
|
|
assert "provision_mol_population_history.py" in dockerfile
|
|
assert "provision_mol_historical_landuse.py" in dockerfile
|
|
assert "fake" not in population.lower()
|
|
|
|
|
|
def test_tower_deploy_waits_for_startup_migration_before_live_smoke() -> None:
|
|
for relative_path in ("scripts/deploy_tower.ps1", "scripts/deploy_tower.sh"):
|
|
script = (ROOT / relative_path).read_text(encoding="utf-8")
|
|
assert "bash deploy/unraid/deploy-release.sh" in script
|
|
|
|
release_script = (ROOT / "deploy/unraid/deploy-release.sh").read_text(encoding="utf-8")
|
|
wait_position = release_script.index("wait_for_geointel_health")
|
|
invocation_position = release_script.index("\n wait_for_geointel_health", wait_position)
|
|
smoke_position = release_script.index("LIVE_SMOKE_CONTAINER=geointel bash scripts/live_migration_smoke.sh")
|
|
assert "docker inspect --format" in release_script
|
|
assert invocation_position < smoke_position
|