diff --git a/backend/migrations/env.py b/backend/migrations/env.py
index affd1d7e2..eaf15930a 100644
--- a/backend/migrations/env.py
+++ b/backend/migrations/env.py
@@ -30,14 +30,13 @@
fileConfig(config.config_file_name)
-# Note: transitional expand/contract exclusions
-# `alembic check` complains that the ORM models don't reference legacy solution-
-# specific fields that were replaced by the `attributes` field. This happens
-# because we are doing expand/contract migrations (for two separate releases);
-# the following release will remove these exceptions.
-_EXPAND_CONTRACT_IGNORED_TABLES: set[str] = {"artefact_bundled_builds_association"}
-_EXPAND_CONTRACT_IGNORED_COLUMNS: set[tuple[str, str]] = {("artefact", "bundled_builds_hash")}
-_EXPAND_CONTRACT_IGNORED_INDEXES: set[str] = {"unique_solution"}
+# Transitional expand/contract exclusions
+# In expand/contract migrations it could happen that the ORM models temporarily
+# don't reference all fields from the database (i.e. in an expand release).
+# `alembic check` will notice that and fail in CI, so we need these exceptions
+_EXPAND_CONTRACT_IGNORED_TABLES: set[str] = set()
+_EXPAND_CONTRACT_IGNORED_COLUMNS: set[tuple[str, str]] = set()
+_EXPAND_CONTRACT_IGNORED_INDEXES: set[str] = set()
def include_object(object, name, type_, reflected, compare_to): # noqa: ANN001, ANN201, ARG001
diff --git a/backend/migrations/versions/2026_08_10_1716-8bd1f5009f02_drop_solution_specific_bundled_build_fields.py b/backend/migrations/versions/2026_08_10_1716-8bd1f5009f02_drop_solution_specific_bundled_build_fields.py
new file mode 100644
index 000000000..2b196e746
--- /dev/null
+++ b/backend/migrations/versions/2026_08_10_1716-8bd1f5009f02_drop_solution_specific_bundled_build_fields.py
@@ -0,0 +1,194 @@
+# Copyright 2026 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see .
+#
+# SPDX-FileCopyrightText: Copyright 2026 Canonical Ltd.
+# SPDX-License-Identifier: AGPL-3.0-only
+
+"""Drop solution-specific bundled build fields
+
+This is the destructive (contract) half of adding the ``attributes`` field
+to artefacts and removing the solution-specific fields.
+It swaps the ``unique_solution`` index to ``(name, version)`` and drops
+``artefact.bundled_builds_hash`` and ``artefact_bundled_builds_association``.
+
+For a safe rolling upgrade this migration must only be deployed *after* the
+expand migration (``8202f7b5953e``) and the code that stops using the old fields
+have been fully rolled out. It re-runs the backfill first so that any rows
+written by not-yet-upgraded units during that rollout (which populate the old
+columns but not ``attributes``) are copied over before the columns are dropped.
+
+Revision ID: 8bd1f5009f02
+Revises: 8202f7b5953e
+Create Date: 2026-08-10 17:16:00.000000+00:00
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "8bd1f5009f02"
+down_revision = "8202f7b5953e"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ # Catch leftover data written by old code during the rollout of the expand
+ # migration before the source columns/table are removed.
+ _copy_bundled_builds_to_attributes()
+ _assert_no_duplicate_solutions(["name", "version"])
+ op.drop_index("unique_solution", table_name="artefact", postgresql_where="(family = 'solution'::familyname)")
+ op.create_index(
+ "unique_solution", "artefact", ["name", "version"], unique=True, postgresql_where=sa.text("family = 'solution'")
+ )
+ op.drop_table("artefact_bundled_builds_association")
+ op.drop_column("artefact", "bundled_builds_hash")
+
+
+def downgrade() -> None:
+ _add_bundled_builds()
+ _assert_no_duplicate_solutions(
+ ["name", "source", "version", "track", "stage", "bundled_builds_hash"],
+ nullable_columns=["bundled_builds_hash"],
+ )
+ op.drop_index("unique_solution", table_name="artefact", postgresql_where=sa.text("family = 'solution'"))
+ op.create_index(
+ "unique_solution",
+ "artefact",
+ ["name", "source", "version", "track", "stage", "bundled_builds_hash"],
+ unique=True,
+ postgresql_where="(family = 'solution'::familyname)",
+ )
+
+
+def _assert_no_duplicate_solutions(key_columns: list[str], nullable_columns: list[str] | None = None) -> None:
+ """Fail fast with a clear error if applying a unique index on ``key_columns`` (scoped to
+ solution artefacts) would violate uniqueness, instead of letting index creation fail with an
+ opaque database error.
+
+ Postgres unique indexes treat NULL as distinct from any other value (including another NULL),
+ so columns listed in ``nullable_columns`` are excluded from the duplicate search whenever they
+ are NULL, matching the semantics of the index we're about to create.
+ """
+ nullable_columns = nullable_columns or []
+ columns_sql = ", ".join(key_columns)
+ not_null_clauses = " AND ".join(f"{column} IS NOT NULL" for column in nullable_columns)
+ where_clause = f"family = 'solution' AND {not_null_clauses}" if not_null_clauses else "family = 'solution'"
+
+ conn = op.get_bind()
+ duplicates = conn.execute(
+ sa.text(f"""
+ SELECT {columns_sql}, COUNT(*) AS duplicate_count
+ FROM artefact
+ WHERE {where_clause}
+ GROUP BY {columns_sql}
+ HAVING COUNT(*) > 1
+ LIMIT 5
+ """) # noqa: S608 - key_columns/nullable_columns are fixed, developer-controlled constants
+ ).fetchall()
+
+ if duplicates:
+ raise RuntimeError(
+ f"Cannot create unique index on solutions ({columns_sql}): found existing duplicate rows "
+ f"(showing up to 5): {duplicates}. Resolve these duplicates manually before running this migration."
+ )
+
+
+def _add_bundled_builds() -> None:
+ op.add_column(
+ "artefact", sa.Column("bundled_builds_hash", sa.VARCHAR(length=64), autoincrement=False, nullable=True)
+ )
+ op.create_table(
+ "artefact_bundled_builds_association",
+ sa.Column("artefact_id", sa.INTEGER(), autoincrement=False, nullable=False),
+ sa.Column("artefact_build_id", sa.INTEGER(), autoincrement=False, nullable=False),
+ sa.ForeignKeyConstraint(
+ ["artefact_build_id"],
+ ["artefact_build.id"],
+ name="artefact_bundled_builds_association_id_fkey",
+ ondelete="CASCADE",
+ ),
+ sa.ForeignKeyConstraint(
+ ["artefact_id"], ["artefact.id"], name="artefact_bundled_builds_artefact_id_fkey", ondelete="CASCADE"
+ ),
+ sa.PrimaryKeyConstraint("artefact_id", "artefact_build_id", name="artefact_bundled_builds_association_pkey"),
+ )
+ _restore_bundled_builds_from_attributes()
+
+
+def _copy_bundled_builds_to_attributes() -> None:
+ op.execute(
+ """
+ UPDATE artefact AS a
+ SET attributes = a.attributes
+ || jsonb_strip_nulls(
+ jsonb_build_object('bundled_builds_hash', a.bundled_builds_hash)
+ )
+ || COALESCE(
+ (
+ SELECT jsonb_build_object(
+ 'bundled_builds',
+ jsonb_agg(assoc.artefact_build_id ORDER BY assoc.artefact_build_id)
+ )
+ FROM artefact_bundled_builds_association assoc
+ WHERE assoc.artefact_id = a.id
+ HAVING count(*) > 0
+ ),
+ '{}'::jsonb
+ )
+ WHERE a.bundled_builds_hash IS NOT NULL
+ OR EXISTS (
+ SELECT 1
+ FROM artefact_bundled_builds_association assoc
+ WHERE assoc.artefact_id = a.id
+ )
+ """
+ )
+
+
+def _restore_bundled_builds_from_attributes() -> None:
+ op.execute(
+ """
+ UPDATE artefact AS a
+ SET bundled_builds_hash = a.attributes ->> 'bundled_builds_hash'
+ WHERE a.attributes ? 'bundled_builds_hash'
+ """
+ )
+ # attributes is API-writable and not schema-validated, so bundled_builds may be a non-array,
+ # contain non-numeric elements, or reference unknown build ids. Guard every step so a malformed
+ # value can never abort the downgrade:
+ # - the CASE feeding jsonb_array_elements_text ensures it only ever sees an array;
+ # - the CASE around ::int only casts digit-only, in-range strings (NULL otherwise, which the
+ # join drops), avoiding scalar-extraction and invalid-cast/overflow errors;
+ # - the join to artefact_build drops ids that don't correspond to a real build (FK safety).
+ op.execute(
+ """
+ INSERT INTO artefact_bundled_builds_association (artefact_id, artefact_build_id)
+ SELECT DISTINCT a.id, ab.id
+ FROM artefact a
+ CROSS JOIN LATERAL jsonb_array_elements_text(
+ CASE
+ WHEN jsonb_typeof(a.attributes -> 'bundled_builds') = 'array'
+ THEN a.attributes -> 'bundled_builds'
+ ELSE '[]'::jsonb
+ END
+ ) AS elem(value)
+ JOIN artefact_build ab
+ ON ab.id = CASE
+ WHEN elem.value ~ '^[0-9]+$'
+ AND length(elem.value) <= 10
+ AND elem.value::bigint <= 2147483647
+ THEN elem.value::int
+ END
+ """
+ )
diff --git a/backend/tests/data_access/test_models.py b/backend/tests/data_access/test_models.py
index 1a332635d..c9b06b3d0 100644
--- a/backend/tests/data_access/test_models.py
+++ b/backend/tests/data_access/test_models.py
@@ -14,9 +14,11 @@
# SPDX-License-Identifier: AGPL-3.0-only
import pytest
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
from test_observer.data_access.models import Issue
-from test_observer.data_access.models_enums import FamilyName, IssueSource
+from test_observer.data_access.models_enums import FamilyName, IssueSource, StageName
from tests.data_generator import DataGenerator
@@ -62,6 +64,38 @@ def test_solutions_with_same_name_and_different_versions_are_allowed(generator:
assert first.id != second.id
+def test_solution_unique_constraint_ignores_source_track_and_stage(
+ generator: DataGenerator, db_session: Session
+) -> None:
+ generator.gen_artefact(
+ family=FamilyName.solution,
+ name="solution",
+ version="1.0",
+ source="first-source",
+ track="first-track",
+ stage=StageName.beta,
+ )
+
+ with pytest.raises(IntegrityError):
+ generator.gen_artefact(
+ family=FamilyName.solution,
+ name="solution",
+ version="1.0",
+ source="second-source",
+ track="second-track",
+ stage=StageName.stable,
+ )
+ db_session.rollback()
+
+
+def test_solutions_with_same_name_and_version_are_unique(generator: DataGenerator, db_session: Session) -> None:
+ generator.gen_artefact(family=FamilyName.solution, name="solution", version="1.0")
+
+ with pytest.raises(IntegrityError):
+ generator.gen_artefact(family=FamilyName.solution, name="solution", version="1.0")
+ db_session.rollback()
+
+
def test_solutions_with_same_version_and_different_names_are_allowed(generator: DataGenerator) -> None:
first = generator.gen_artefact(family=FamilyName.solution, name="solution-a", version="1.0")
second = generator.gen_artefact(family=FamilyName.solution, name="solution-b", version="1.0")
diff --git a/backend/tests/migrations/test_8bd1f5009f02_drop_solution_specific_bundled_build_fields.py b/backend/tests/migrations/test_8bd1f5009f02_drop_solution_specific_bundled_build_fields.py
new file mode 100644
index 000000000..38bfb08a5
--- /dev/null
+++ b/backend/tests/migrations/test_8bd1f5009f02_drop_solution_specific_bundled_build_fields.py
@@ -0,0 +1,435 @@
+# Copyright 2026 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see .
+#
+# SPDX-FileCopyrightText: Copyright 2026 Canonical Ltd.
+# SPDX-License-Identifier: AGPL-3.0-only
+
+"""Tests for the contract half: dropping the solution-specific bundled build fields.
+
+This migration swaps the ``unique_solution`` index to ``(name, version)`` and
+drops ``artefact.bundled_builds_hash`` and ``artefact_bundled_builds_association``.
+It re-runs the backfill first so rows written by not-yet-upgraded units during
+the expand rollout are copied into ``attributes`` before the columns are dropped.
+
+``PREVIOUS_REV`` is the expand migration (attributes already added and backfilled).
+"""
+
+from collections.abc import Generator
+from urllib.parse import urlparse, urlunparse
+
+import pytest
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import Engine, create_engine, text
+from sqlalchemy.engine import Connection
+from sqlalchemy_utils import create_database, database_exists, drop_database # type: ignore[import-untyped]
+
+# The expand migration (adds + backfills attributes; old column/table still present).
+PREVIOUS_REV = "8202f7b5953e"
+# The contract migration under test (drops old column/table, swaps index).
+TARGET_REV = "8bd1f5009f02"
+
+
+@pytest.fixture
+def migration_context(db_url: str) -> Generator[tuple[Engine, Config], None, None]:
+ parsed = urlparse(db_url)
+ test_db_url = urlunparse(
+ (parsed.scheme, parsed.netloc, "/test_migration_drop_bundled", parsed.params, parsed.query, parsed.fragment)
+ )
+
+ if database_exists(test_db_url):
+ drop_database(test_db_url)
+
+ create_database(test_db_url)
+
+ engine: Engine | None = None
+ try:
+ engine = create_engine(test_db_url)
+ alembic_config = Config("alembic.ini")
+ alembic_config.set_main_option("sqlalchemy.url", test_db_url)
+
+ yield engine, alembic_config
+ finally:
+ if engine is not None:
+ engine.dispose()
+ if database_exists(test_db_url):
+ drop_database(test_db_url)
+
+
+def _insert_legacy_artefact(
+ conn: Connection,
+ name: str,
+ bundled_builds_hash: str | None = None,
+ version: str = "1.0",
+ track: str = "latest",
+ source: str = "source",
+ stage: str = "stable",
+) -> int:
+ """Insert a solution artefact at the expand revision, where both the ``attributes``
+ column and the legacy ``bundled_builds_hash`` column exist. ``attributes`` is left at
+ its server default ({}) to simulate rows written by not-yet-upgraded code."""
+ result = conn.execute(
+ text("""
+ INSERT INTO artefact (
+ name, version, stage, family, status, archived, bug_link, comment,
+ store, branch, track, series, repo, source, os, release, sha256, owner, image_url,
+ created_at, updated_at, bundled_builds_hash
+ )
+ VALUES (
+ :name, :version, :stage, 'solution', 'UNDECIDED', false, '', '',
+ '', '', :track, '', '', :source, '', '', '', '', '',
+ NOW(), NOW(), :bundled_builds_hash
+ )
+ RETURNING id
+ """),
+ {
+ "name": name,
+ "version": version,
+ "stage": stage,
+ "track": track,
+ "source": source,
+ "bundled_builds_hash": bundled_builds_hash,
+ },
+ )
+ return result.scalar_one()
+
+
+def _insert_artefact(
+ conn: Connection,
+ name: str,
+ attributes: str = "{}",
+ version: str = "1.0",
+ track: str = "latest",
+ source: str = "source",
+ stage: str = "stable",
+) -> int:
+ """Insert a solution artefact at the contracted revision, where only the ``attributes``
+ column exists (``bundled_builds_hash`` has been dropped)."""
+ result = conn.execute(
+ text("""
+ INSERT INTO artefact (
+ name, version, stage, family, status, archived, bug_link, comment,
+ store, branch, track, series, repo, source, os, release, sha256, owner, image_url,
+ created_at, updated_at, attributes
+ )
+ VALUES (
+ :name, :version, :stage, 'solution', 'UNDECIDED', false, '', '',
+ '', '', :track, '', '', :source, '', '', '', '', '',
+ NOW(), NOW(), CAST(:attributes AS jsonb)
+ )
+ RETURNING id
+ """),
+ {
+ "name": name,
+ "version": version,
+ "stage": stage,
+ "track": track,
+ "source": source,
+ "attributes": attributes,
+ },
+ )
+ return result.scalar_one()
+
+
+def _insert_artefact_build(conn: Connection, artefact_id: int, architecture: str = "amd64") -> int:
+ result = conn.execute(
+ text("""
+ INSERT INTO artefact_build (architecture, revision, artefact_id, created_at, updated_at)
+ VALUES (:architecture, NULL, :artefact_id, NOW(), NOW())
+ RETURNING id
+ """),
+ {"architecture": architecture, "artefact_id": artefact_id},
+ )
+ return result.scalar_one()
+
+
+def _insert_association(conn: Connection, artefact_id: int, artefact_build_id: int) -> None:
+ conn.execute(
+ text("""
+ INSERT INTO artefact_bundled_builds_association (artefact_id, artefact_build_id)
+ VALUES (:artefact_id, :artefact_build_id)
+ """),
+ {"artefact_id": artefact_id, "artefact_build_id": artefact_build_id},
+ )
+
+
+def _attribute_text(engine: Engine, artefact_id: int, key: str) -> str | None:
+ with engine.connect() as conn:
+ return conn.execute(
+ text("SELECT attributes::jsonb ->> :key FROM artefact WHERE id = :artefact_id"),
+ {"artefact_id": artefact_id, "key": key},
+ ).scalar_one()
+
+
+def _bundled_build_ids(engine: Engine, artefact_id: int) -> list[int]:
+ with engine.connect() as conn:
+ return list(
+ conn.execute(
+ text("""
+ SELECT jsonb_array_elements_text(attributes::jsonb -> 'bundled_builds')::int
+ FROM artefact
+ WHERE id = :artefact_id
+ """),
+ {"artefact_id": artefact_id},
+ ).scalars()
+ )
+
+
+def test_upgrade_backfills_leftovers_before_dropping(migration_context: tuple[Engine, Config]) -> None:
+ """A row written by not-yet-upgraded code (legacy columns set, attributes still empty)
+ after the expand migration must be copied into attributes before the columns are dropped."""
+ engine, alembic_config = migration_context
+ command.upgrade(alembic_config, PREVIOUS_REV)
+ with engine.begin() as conn:
+ artefact_id = _insert_legacy_artefact(conn, "solution-leftover", bundled_builds_hash="hash-value")
+ build_id = _insert_artefact_build(conn, artefact_id)
+ _insert_association(conn, artefact_id, build_id)
+
+ command.upgrade(alembic_config, TARGET_REV)
+
+ assert _attribute_text(engine, artefact_id, "bundled_builds_hash") == "hash-value"
+ assert _bundled_build_ids(engine, artefact_id) == [build_id]
+
+
+def test_downgrade_restores_empty_attributes(migration_context: tuple[Engine, Config]) -> None:
+ engine, alembic_config = migration_context
+ command.upgrade(alembic_config, TARGET_REV)
+ with engine.begin() as conn:
+ artefact_id = _insert_artefact(conn, "solution-downgrade-empty", attributes="{}")
+
+ command.downgrade(alembic_config, PREVIOUS_REV)
+
+ with engine.connect() as conn:
+ bundled_hash = conn.execute(
+ text("SELECT bundled_builds_hash FROM artefact WHERE id = :id"),
+ {"id": artefact_id},
+ ).scalar_one()
+ association_count = conn.execute(
+ text("SELECT count(*) FROM artefact_bundled_builds_association WHERE artefact_id = :id"),
+ {"id": artefact_id},
+ ).scalar_one()
+ assert bundled_hash is None
+ assert association_count == 0
+
+
+def test_downgrade_restores_hash_and_associations(migration_context: tuple[Engine, Config]) -> None:
+ engine, alembic_config = migration_context
+ command.upgrade(alembic_config, TARGET_REV)
+ with engine.begin() as conn:
+ artefact_id = _insert_artefact(conn, "solution-downgrade-both", attributes="{}")
+ first_build_id = _insert_artefact_build(conn, artefact_id, architecture="amd64")
+ second_build_id = _insert_artefact_build(conn, artefact_id, architecture="arm64")
+ conn.execute(
+ text("""
+ UPDATE artefact
+ SET attributes = jsonb_build_object(
+ 'bundled_builds_hash', 'restored-hash',
+ 'bundled_builds', jsonb_build_array(CAST(:first_build_id AS int), CAST(:second_build_id AS int))
+ )
+ WHERE id = :artefact_id
+ """),
+ {
+ "artefact_id": artefact_id,
+ "first_build_id": first_build_id,
+ "second_build_id": second_build_id,
+ },
+ )
+
+ command.downgrade(alembic_config, PREVIOUS_REV)
+
+ with engine.connect() as conn:
+ bundled_hash = conn.execute(
+ text("SELECT bundled_builds_hash FROM artefact WHERE id = :id"),
+ {"id": artefact_id},
+ ).scalar_one()
+ association_ids = list(
+ conn.execute(
+ text("""
+ SELECT artefact_build_id
+ FROM artefact_bundled_builds_association
+ WHERE artefact_id = :id
+ ORDER BY artefact_build_id
+ """),
+ {"id": artefact_id},
+ ).scalars()
+ )
+ assert bundled_hash == "restored-hash"
+ assert association_ids == [first_build_id, second_build_id]
+
+
+def test_downgrade_tolerates_malformed_bundled_builds(migration_context: tuple[Engine, Config]) -> None:
+ """``attributes`` is writable via the API and is not schema-validated, so ``bundled_builds`` can
+ hold arbitrary JSON (a non-array value, non-numeric elements, or unknown build ids). The
+ downgrade must not blow up with a Postgres JSON/cast/FK error - it should skip invalid data and
+ still restore the valid parts, so rollback is always possible."""
+ engine, alembic_config = migration_context
+ command.upgrade(alembic_config, TARGET_REV)
+ with engine.begin() as conn:
+ # A real build so we can prove valid ids are still restored alongside the bad ones.
+ valid_artefact_id = _insert_artefact(conn, "solution-valid", attributes="{}")
+ valid_build_id = _insert_artefact_build(conn, valid_artefact_id)
+ conn.execute(
+ text("""
+ UPDATE artefact
+ SET attributes = jsonb_build_object(
+ 'bundled_builds_hash', 'keep-hash',
+ 'bundled_builds', jsonb_build_array(CAST(:valid_build_id AS int))
+ )
+ WHERE id = :artefact_id
+ """),
+ {"artefact_id": valid_artefact_id, "valid_build_id": valid_build_id},
+ )
+
+ # bundled_builds is a scalar string, not an array -> jsonb_array_elements_text fails.
+ non_array_id = _insert_artefact(conn, "solution-non-array", attributes='{"bundled_builds": "not-an-array"}')
+ # bundled_builds contains a non-numeric element -> ::int cast fails.
+ non_numeric_id = _insert_artefact(
+ conn, "solution-non-numeric", attributes='{"bundled_builds": ["not-a-number"]}'
+ )
+ # bundled_builds references a build id that does not exist -> FK violation.
+ unknown_build_id = _insert_artefact(
+ conn, "solution-unknown-build", attributes='{"bundled_builds": [999999999]}'
+ )
+
+ # Raises a DatabaseError; after guarding the traversal it should complete cleanly.
+ command.downgrade(alembic_config, PREVIOUS_REV)
+
+ with engine.connect() as conn:
+ # Valid data is still restored.
+ assert (
+ conn.execute(
+ text("SELECT bundled_builds_hash FROM artefact WHERE id = :id"), {"id": valid_artefact_id}
+ ).scalar_one()
+ == "keep-hash"
+ )
+ assert list(
+ conn.execute(
+ text("SELECT artefact_build_id FROM artefact_bundled_builds_association WHERE artefact_id = :id"),
+ {"id": valid_artefact_id},
+ ).scalars()
+ ) == [valid_build_id]
+
+ # Malformed entries produce no association rows rather than aborting the whole downgrade.
+ for bad_id in (non_array_id, non_numeric_id, unknown_build_id):
+ assert (
+ conn.execute(
+ text("SELECT count(*) FROM artefact_bundled_builds_association WHERE artefact_id = :id"),
+ {"id": bad_id},
+ ).scalar_one()
+ == 0
+ )
+
+
+def test_upgrade_fails_fast_on_duplicate_name_and_version(migration_context: tuple[Engine, Config]) -> None:
+ """The expand-revision schema still allows several solutions sharing (name, version) as long as
+ track/source differ; the contract upgrade must refuse to create the tighter (name, version)
+ unique index rather than fail with an opaque database error, and must not partially apply."""
+ engine, alembic_config = migration_context
+ command.upgrade(alembic_config, PREVIOUS_REV)
+ with engine.begin() as conn:
+ _insert_legacy_artefact(conn, "dup-solution", version="1.0", track="track-a", source="source-a")
+ _insert_legacy_artefact(conn, "dup-solution", version="1.0", track="track-b", source="source-b")
+
+ with pytest.raises(RuntimeError, match="Cannot create unique index"):
+ command.upgrade(alembic_config, TARGET_REV)
+
+ # The failed migration must not have dropped the legacy column.
+ with engine.connect() as conn:
+ bundled_hash_column = conn.execute(
+ text("""
+ SELECT column_name
+ FROM information_schema.columns
+ WHERE table_name = 'artefact' AND column_name = 'bundled_builds_hash'
+ """)
+ ).fetchone()
+ assert bundled_hash_column is not None
+
+
+def test_downgrade_fails_fast_on_duplicate_widened_key(migration_context: tuple[Engine, Config]) -> None:
+ """Defends the downgrade's wider unique index the same way, in case data ever ends up violating
+ it (e.g. the (name, version) index was bypassed or dropped out-of-band)."""
+ engine, alembic_config = migration_context
+ command.upgrade(alembic_config, TARGET_REV)
+ with engine.begin() as conn:
+ conn.execute(text("DROP INDEX unique_solution"))
+ _insert_artefact(conn, "dup-solution", attributes='{"bundled_builds_hash": "hash-x"}')
+ _insert_artefact(conn, "dup-solution", attributes='{"bundled_builds_hash": "hash-x"}')
+
+ with pytest.raises(RuntimeError, match="Cannot create unique index"):
+ command.downgrade(alembic_config, PREVIOUS_REV)
+
+
+def test_upgrade_schema_changes(migration_context: tuple[Engine, Config]) -> None:
+ engine, alembic_config = migration_context
+ command.upgrade(alembic_config, TARGET_REV)
+
+ with engine.connect() as conn:
+ attributes_column = conn.execute(
+ text("""
+ SELECT column_name
+ FROM information_schema.columns
+ WHERE table_name = 'artefact' AND column_name = 'attributes'
+ """)
+ ).fetchone()
+ association_table = conn.execute(
+ text("""
+ SELECT table_name
+ FROM information_schema.tables
+ WHERE table_name = 'artefact_bundled_builds_association'
+ """)
+ ).fetchone()
+ bundled_hash_column = conn.execute(
+ text("""
+ SELECT column_name
+ FROM information_schema.columns
+ WHERE table_name = 'artefact' AND column_name = 'bundled_builds_hash'
+ """)
+ ).fetchone()
+
+ assert attributes_column is not None
+ assert association_table is None
+ assert bundled_hash_column is None
+
+
+def test_downgrade_schema_changes(migration_context: tuple[Engine, Config]) -> None:
+ engine, alembic_config = migration_context
+ command.upgrade(alembic_config, TARGET_REV)
+ command.downgrade(alembic_config, PREVIOUS_REV)
+
+ with engine.connect() as conn:
+ attributes_column = conn.execute(
+ text("""
+ SELECT column_name
+ FROM information_schema.columns
+ WHERE table_name = 'artefact' AND column_name = 'attributes'
+ """)
+ ).fetchone()
+ association_table = conn.execute(
+ text("""
+ SELECT table_name
+ FROM information_schema.tables
+ WHERE table_name = 'artefact_bundled_builds_association'
+ """)
+ ).fetchone()
+ bundled_hash_column = conn.execute(
+ text("""
+ SELECT column_name
+ FROM information_schema.columns
+ WHERE table_name = 'artefact' AND column_name = 'bundled_builds_hash'
+ """)
+ ).fetchone()
+
+ # Downgrading the contract migration restores the old schema but keeps attributes
+ # (attributes is removed only by downgrading the expand migration).
+ assert attributes_column is not None
+ assert association_table is not None
+ assert bundled_hash_column is not None