test: prepare isolated PostgreSQL publication integration checks

README.md

@@ -141,8 +141,10 @@ node --check app/static/admin.js
 
 Тесты с данными используют временную SQLite; отдельные API smoke-тесты запускают приложение с его текущей конфигурацией. Браузерная проверка: `pip install playwright`, `python -m playwright install chromium`, затем `python tests/browser_admin.py`. Она проверяет реальный рендеринг страниц, клавиатуру, диалог, фильтры и восстановление прогресса после ошибки. Скриншоты сохраняются во временную папку; путь выводится в конце.
 
-Версия сервиса: `0.10.2`. Админка всегда показывает версии backend и frontend в footer.
+Версия сервиса: `0.10.3`. Админка всегда показывает версии backend и frontend в footer.
 
 ## Разовый импорт публикаций
 
 Инструкция, минимальные права временной роли и ограниченный SSH-запуск: [docs/publication-import.md](docs/publication-import.md). По умолчанию импортер выполняет dry-run; production-применение требует отдельного решения владельца.
+
+Интеграционная проверка импорта на отдельном PostgreSQL: [docs/postgresql-import-verification.md](docs/postgresql-import-verification.md).

app/version.py

@@ -1,3 +1,3 @@
-APP_VERSION = "0.10.2"
-FRONTEND_VERSION = "0.10.2"
-BACKEND_VERSION = "0.10.2"
+APP_VERSION = "0.10.3"
+FRONTEND_VERSION = "0.10.3"
+BACKEND_VERSION = "0.10.3"

docs/postgresql-import-verification.md

@@ -0,0 +1,96 @@
+# Проверка импорта на изолированном PostgreSQL
+
+Запускается средствами проекта: pytest, SQLAlchemy, psycopg и отдельный Docker
+Compose. Не использует production Compose, `.env`, SSH или данные сотрудников.
+PostgreSQL 16 и runner работают в отдельной `internal`-сети без опубликованных
+портов. Данные PostgreSQL находятся в tmpfs, без постоянных volumes. Базы:
+`miem_import_test` и `miem_crawl_test`. Все сотрудники, статьи и credentials
+синтетические; тестовые пароли непригодны для реальных систем.
+
+Dockerfile копирует только нужный код и requirements. Его отдельный dockerignore
+исключает `.git`, `.env`, файлы ключей/credentials, CSV владельца, данные
+`app/data` и остальные файлы workspace. Runtime не получает переменные хоста.
+CLI subprocess дополнительно получает только явно заданные тестовые libpq
+параметры; `.pgpass`/service-файлы отключены. Не изменяйте адрес `postgres-test`
+или имена тестовых БД для подключения к существующему серверу.
+
+Синтетическая роль `import_test` создаётся только init-скриптом одноразового
+тестового кластера. Fixture выдаёт ей документированные column grants и USAGE
+sequence. Проверяется отсутствие CREATE схемы, INSERT сотрудников, DELETE
+публикаций и sequence UPDATE. Это не создание временного доступа к рабочей БД.
+Таблицы создаются моделью приложения и очищаются только в тестовых БД.
+
+HTTP заблокирован test fixture; входные профили полного обхода подставляются
+локально. Это проверяет настоящий SQL, транзакции, savepoint и сериализацию JSON,
+но не парсинг сайта ВШЭ. Fixtures SQLite и подмена psycopg для интеграционных
+тестов не используются.
+
+## Что проверяется
+
+1. Dry-run: все строки всех таблиц и состояния sequences совпадают до/после.
+2. Apply: одна вставка, одно обновление, пропуск отсутствующего сотрудника.
+3. Повторный импорт: ноль INSERT/UPDATE, неизменные таблицы/sequences.
+4. CHECK constraint тестовой БД вызывает ошибку третьего действия после UPDATE
+   и успешного INSERT; все строки возвращаются в исходное состояние. Sequence
+   может продвинуться при rollback — это ожидаемое поведение PostgreSQL.
+5. Обновление только года без raw_data сохраняет существующий JSONB.
+6. Поздние строки CSV дополняют метаданные; первое непустое значение сохраняется
+   при конфликте, отчёт содержит названия конфликтующих полей. Повтор идемпотентен.
+7. Два последовательных crawl сохраняют CSV-статью, отсутствующую у ВШЭ;
+   появившаяся статья получает HSE ID без второй записи и потери метаданных.
+8. Полный `run_crawl` обрабатывает двух сотрудников. Два HSE ID на одну CSV-запись
+   откатывают savepoint первого, сохраняя JSONB, авторов, ID, хеш и updated_at;
+   публикация второго сохраняется. Проверяются обе очередности ID и уже
+   установленная привязка; неоднозначность отражается в CrawlError.
+
+Всего 10 PostgreSQL-тестов (восьмой сценарий параметризован). В обычном
+`python -m pytest -q` они пропускаются. Opt-in `MIEM_POSTGRES_INTEGRATION=1`
+разрешён только с точным тестовым DATABASE_URL Compose-проекта.
+
+## Команды владельцу
+
+Выполняйте на локальной машине с Docker Engine/Compose v2 из корня репозитория
+после независимого ревью. Команды явно выбирают локальный Unix socket и очищают
+наследуемое окружение. Для Docker Desktop/Windows используйте Linux/WSL с
+доступным локальным `/var/run/docker.sock`; не подменяйте его удалённым VPS.
+Первый шаг:
+
+```sh
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock version
+```
+
+Ожидается exit code 0 и секции `Client` и `Server`. `command not found`,
+`permission denied` или только Client с ошибкой подключения означает, что
+до запуска тестов нужно восстановить локальный Docker; не подключайтесь к VPS.
+
+Дальнейшие точные команды сохранены здесь для воспроизводимого запуска:
+
+```sh
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock compose --env-file /dev/null --project-name miem-publication-integration -f tests/postgres/compose.yml config --quiet
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock compose --env-file /dev/null --project-name miem-publication-integration -f tests/postgres/compose.yml build tests
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock compose --env-file /dev/null --project-name miem-publication-integration -f tests/postgres/compose.yml up --abort-on-container-exit --exit-code-from tests
+```
+
+Первые две команды проверяют Compose и собирают runner с уже используемыми
+зависимостями. Сборка/скачивание образов требуют registry/PyPI, но тестовая
+runtime-сеть не имеет внешнего доступа и запросов к ВШЭ нет. При успехе третьей
+команды ожидаются `10 passed` и exit code 0. Фактическую совместимость образа,
+column grants и PostgreSQL можно считать подтверждённой **только после такого
+выполнения**, не по unit-тестам или статическому чтению файлов.
+
+После сохранения вывода тестов удалите именно этот тестовый проект:
+
+```sh
+env -i PATH="$PATH" docker --host unix:///var/run/docker.sock compose --env-file /dev/null --project-name miem-publication-integration -f tests/postgres/compose.yml down
+```
+
+Это уничтожит только синтетические данные в tmpfs тестового PostgreSQL. Не
+используйте `docker system prune`, production Compose или удаление рабочих
+volumes. Перед повторным прогоном выполните этот же `down`, чтобы следующий
+кластер начинался с чистого init.sql; не запускайте два прогона одновременно.
+
+## Статус подготовки
+
+В среде подготовки Docker отсутствует; настоящие PostgreSQL-тесты не запускались.
+Локально доступны только unit-тесты, проверка Python и статические проверки.
+Production/VPS и доступ к рабочей БД не изменялись.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "miem-workers"
-version = "0.10.2"
+version = "0.10.3"
 description = "MIEM employees parser, admin API, and web admin"
 requires-python = ">=3.11"
 dependencies = [

tests/postgres/Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3.12-slim
+WORKDIR /app
+COPY requirements.txt .
+RUN python -m pip install --no-cache-dir -r requirements.txt
+COPY app ./app
+COPY scripts/import_publications.py ./scripts/import_publications.py
+COPY tests/conftest.py tests/test_publication_import_postgres.py ./tests/

tests/postgres/Dockerfile.dockerignore

@@ -0,0 +1,18 @@
+**
+!requirements.txt
+!app/
+!app/**
+!scripts/
+!scripts/import_publications.py
+!tests/
+!tests/conftest.py
+!tests/test_publication_import_postgres.py
+!tests/postgres/
+!tests/postgres/Dockerfile
+!tests/postgres/Dockerfile.dockerignore
+app/data/
+**/__pycache__/
+**/.env*
+**/*credentials*
+**/*.key
+**/*.pem

tests/postgres/compose.yml

@@ -0,0 +1,36 @@
+name: miem-publication-integration
+
+services:
+  postgres-test:
+    image: postgres:16-alpine
+    environment:
+      POSTGRES_USER: test_admin
+      POSTGRES_PASSWORD: isolated_test_admin
+      POSTGRES_DB: miem_import_test
+    tmpfs:
+      - /var/lib/postgresql/data
+    volumes:
+      - ./init.sql:/docker-entrypoint-initdb.d/10-test.sql:ro
+    healthcheck:
+      test: ["CMD", "pg_isready", "-h", "127.0.0.1", "-U", "test_admin", "-d", "miem_import_test"]
+      interval: 2s
+      timeout: 5s
+      retries: 30
+    networks: [isolated]
+
+  tests:
+    build:
+      context: ../..
+      dockerfile: tests/postgres/Dockerfile
+    environment:
+      DATABASE_URL: postgresql+psycopg://test_admin:isolated_test_admin@postgres-test:5432/miem_import_test
+      MIEM_POSTGRES_INTEGRATION: "1"
+    command: ["python", "-m", "pytest", "-q", "tests/test_publication_import_postgres.py"]
+    depends_on:
+      postgres-test:
+        condition: service_healthy
+    networks: [isolated]
+
+networks:
+  isolated:
+    internal: true

tests/postgres/init.sql

@@ -0,0 +1,6 @@
+-- Synthetic credentials for this ephemeral, private test cluster only.
+CREATE DATABASE miem_crawl_test OWNER test_admin;
+CREATE ROLE import_test LOGIN PASSWORD 'isolated_test_import'
+    NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT NOREPLICATION NOBYPASSRLS
+    CONNECTION LIMIT 1;
+GRANT CONNECT ON DATABASE miem_import_test, miem_crawl_test TO import_test;

tests/test_api.py

@@ -18,7 +18,7 @@ def test_health_returns_versions():
     response = TestClient(app).get("/api/health")
 
     assert response.status_code == 200
-    assert response.json()["backend_version"] == "0.10.2"
+    assert response.json()["backend_version"] == "0.10.3"
 
 
 def test_api_employees_and_stats_require_admin_session():

tests/test_publication_import_postgres.py

@@ -0,0 +1,280 @@
+"""Real PostgreSQL checks; opt in only inside tests/postgres/compose.yml."""
+
+import copy
+import csv
+import json
+import os
+import subprocess
+import sys
+from contextlib import contextmanager
+from pathlib import Path
+
+import pytest
+import requests
+from sqlalchemy import create_engine, select, text
+from sqlalchemy.orm import Session
+
+from app.config import Settings
+from app.db import Base
+from app.models import CrawlError, CrawlRun, Employee, EmployeePublication
+from app.services.crawler import _try_sync_employee_publications, run_crawl
+
+
+ADMIN_URL = "postgresql+psycopg://test_admin:isolated_test_admin@postgres-test:5432/miem_import_test"
+pytestmark = pytest.mark.skipif(
+    os.environ.get("MIEM_POSTGRES_INTEGRATION") != "1", reason="Requires isolated PostgreSQL Compose project"
+)
+
+
+@pytest.fixture(autouse=True)
+def isolated_environment(monkeypatch):
+    assert os.environ.get("DATABASE_URL") == ADMIN_URL, "Use the dedicated test Compose project"
+
+    def forbidden_request(*args, **kwargs):
+        raise AssertionError("External HTTP requests are forbidden in PostgreSQL integration tests")
+
+    monkeypatch.setattr(requests.sessions.Session, "request", forbidden_request)
+
+
+@contextmanager
+def database(name):
+    assert name in {"miem_import_test", "miem_crawl_test"}
+    engine = create_engine(ADMIN_URL.replace("/miem_import_test", "/" + name))
+    try:
+        with engine.connect() as connection:
+            assert connection.scalar(text("SELECT current_database()")) == name
+        Base.metadata.create_all(engine)
+        with engine.begin() as connection:
+            connection.execute(text("GRANT USAGE ON SCHEMA public TO import_test"))
+            connection.execute(text("GRANT SELECT (id, full_name) ON public.employees TO import_test"))
+            connection.execute(text("GRANT SELECT (id, employee_id, title, year, doi_url, raw_data) ON public.employee_publications TO import_test"))
+            connection.execute(text("GRANT INSERT (employee_id, title, year, doi_url, raw_data, source_hash, created_at, updated_at) ON public.employee_publications TO import_test"))
+            connection.execute(text("GRANT UPDATE (title, year, doi_url, raw_data, updated_at) ON public.employee_publications TO import_test"))
+            # Created from the actual model: not an assumed sequence name.
+            sequence = connection.scalar(text("SELECT pg_get_serial_sequence('public.employee_publications', 'id')"))
+            assert sequence == "public.employee_publications_id_seq"
+            connection.execute(text("GRANT USAGE ON SEQUENCE public.employee_publications_id_seq TO import_test"))
+            permissions = connection.execute(text("""SELECT
+                has_schema_privilege('import_test', 'public', 'CREATE'),
+                has_table_privilege('import_test', 'public.employees', 'INSERT'),
+                has_table_privilege('import_test', 'public.employee_publications', 'DELETE'),
+                has_sequence_privilege('import_test', 'public.employee_publications_id_seq', 'UPDATE')
+            """)).one()
+            assert not any(permissions)
+        with Session(engine) as session:
+            yield session, name
+    finally:
+        # Only the dedicated ephemeral test DB; never DROP DATABASE or roles.
+        Base.metadata.drop_all(engine)
+        engine.dispose()
+
+
+@pytest.fixture
+def pg_import_db():
+    with database("miem_import_test") as db:
+        yield db
+
+
+@pytest.fixture
+def pg_crawl_db():
+    with database("miem_crawl_test") as db:
+        yield db
+
+
+def employee(session, name="Synthetic One", key="one"):
+    record = Employee(full_name=name, profile_key="staff:" + key,
+                      canonical_url="https://synthetic.invalid/staff/" + key)
+    session.add(record)
+    session.commit()
+    return record
+
+
+def article(title="Synthetic article", identifier="10.1234/test", year="2024", author="Synthetic One", **metadata):
+    return {"Название публикации": title, "Год": year, "DOI": identifier,
+            "Авторы НИУ ВШЭ": author, **metadata}
+
+
+def write_csv(path, rows):
+    columns = list(dict.fromkeys(key for row in rows for key in row))
+    with path.open("w", encoding="utf-8-sig", newline="") as output:
+        writer = csv.DictWriter(output, fieldnames=columns, delimiter=";")
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def cli(tmp_path, db, rows, *, apply=False, expected_code=0):
+    _, name = db
+    source = tmp_path / "articles.csv"
+    report = tmp_path / "report.json"
+    write_csv(source, rows)
+    command = [sys.executable, str(Path(__file__).resolve().parents[1] / "scripts/import_publications.py"),
+               str(source), "--report", str(report)]
+    if apply:
+        command.append("--apply")
+    # Explicit whitelist: no production env, .env, pg_service or ~/.pgpass.
+    environment = {"PATH": os.environ["PATH"], "LANG": "C.UTF-8", "PGHOST": "postgres-test",
+                   "PGPORT": "5432", "PGDATABASE": name, "PGUSER": "import_test",
+                   "PGPASSWORD": "isolated_test_import", "PGPASSFILE": "/dev/null",
+                   "PGSERVICEFILE": "/dev/null", "PGCONNECT_TIMEOUT": "5"}
+    result = subprocess.run(command, env=environment, cwd=tmp_path, capture_output=True, text=True, timeout=60)
+    assert result.returncode == expected_code, result.stdout + result.stderr
+    assert "Synthetic One" not in result.stdout and "isolated_test_import" not in result.stdout
+    return json.loads(report.read_text()) if report.exists() else None
+
+
+def snapshot(session):
+    session.expire_all()
+    tables = {table.name: session.execute(select(table).order_by(*table.primary_key.columns)).all()
+              for table in Base.metadata.sorted_tables}
+    sequences = session.execute(text("SELECT sequencename, last_value FROM pg_sequences WHERE schemaname = 'public' ORDER BY sequencename")).all()
+    session.commit()
+    return tables, sequences
+
+
+def test_postgres_dry_run_changes_neither_rows_nor_sequences(pg_import_db, tmp_path):
+    session, _ = pg_import_db
+    employee(session)
+    before = snapshot(session)
+    report = cli(tmp_path, pg_import_db, [article()])
+    assert report["summary"]["counts"] == {"insert": 1, "update": 0, "skip": 0}
+    assert snapshot(session) == before
+
+
+def test_postgres_apply_and_reimport(pg_import_db, tmp_path):
+    session, _ = pg_import_db
+    person = employee(session)
+    existing = EmployeePublication(employee_id=person.id, title="Existing", year=2024,
+                                   source_hash="existing", raw_data={"hse": "keep"})
+    session.add(existing)
+    session.commit()
+    rows = [article(title="Existing"), article(title="New", identifier="10.1234/new"),
+            article(title="Unknown author", identifier="10.1234/unknown", author="Synthetic Missing")]
+    report = cli(tmp_path, pg_import_db, rows, apply=True)
+    assert report["summary"]["counts"] == {"insert": 1, "update": 1, "skip": 1}
+    session.expire_all()
+    assert session.query(Employee).count() == 1
+    assert session.query(EmployeePublication).count() == 2
+    assert existing.doi_url == "https://doi.org/10.1234/test" and existing.raw_data["hse"] == "keep"
+    before = snapshot(session)
+    repeated = cli(tmp_path, pg_import_db, rows, apply=True)
+    assert repeated["summary"]["counts"] == {"insert": 0, "update": 0, "skip": 3}
+    assert snapshot(session) == before
+
+
+def test_postgres_apply_failure_rolls_back_update_and_insert(pg_import_db, tmp_path):
+    session, _ = pg_import_db
+    person = employee(session)
+    existing = EmployeePublication(employee_id=person.id, title="Existing", year=2024,
+                                   source_hash="existing", raw_data={"hse": "keep"})
+    session.add(existing)
+    session.commit()
+    # Test-only constraint forces a real server error after update + successful INSERT.
+    session.execute(text("ALTER TABLE public.employee_publications ADD CONSTRAINT test_failure CHECK (title <> 'Reject')"))
+    session.commit()
+    before, _ = snapshot(session)
+    rows = [article(title="Existing"), article(title="Accepted", identifier="10.1234/accepted"),
+            article(title="Reject", identifier="10.1234/reject")]
+    assert cli(tmp_path, pg_import_db, rows, apply=True, expected_code=1) is None
+    after, _ = snapshot(session)
+    assert after == before  # nextval may advance on rollback; rows must not change.
+    assert (tmp_path / "report.json.pending").exists()
+
+
+def test_postgres_year_only_update_preserves_json(pg_import_db, tmp_path):
+    session, _ = pg_import_db
+    person = employee(session)
+    row = article()
+    raw = {"hse": {"keep": True}, "csv_enrichment": row.copy()}
+    record = EmployeePublication(employee_id=person.id, title=row["Название публикации"],
+                                 year=None, doi_url="https://doi.org/10.1234/test",
+                                 source_hash="existing", raw_data=raw)
+    session.add(record)
+    session.commit()
+    report = cli(tmp_path, pg_import_db, [row], apply=True)
+    session.expire_all()
+    assert report["summary"]["counts"]["update"] == 1
+    assert record.year == 2024 and record.raw_data == raw
+
+
+def test_postgres_late_csv_metadata_and_conflicts(pg_import_db, tmp_path):
+    session, _ = pg_import_db
+    employee(session)
+    rows = [article(), article(**{"EID (Scopus)": "late-id", "Квартиль scopus": "Q1"}),
+            article(**{"EID (Scopus)": "conflict-id", "UT (WoS)": "late-wos"})]
+    report = cli(tmp_path, pg_import_db, rows, apply=True)
+    session.expire_all()
+    assert session.query(EmployeePublication).count() == 1
+    metadata = session.query(EmployeePublication).one().raw_data["csv_enrichment"]
+    assert metadata["EID (Scopus)"] == "late-id" and metadata["Квартиль scopus"] == "Q1"
+    assert metadata["UT (WoS)"] == "late-wos"
+    assert report["summary"]["metadata_conflicts"] == 1
+    assert report["records"][2]["conflict_fields"] == ["EID (Scopus)"]
+    before = snapshot(session)
+    repeated = cli(tmp_path, pg_import_db, rows, apply=True)
+    assert repeated["summary"]["counts"]["insert"] == repeated["summary"]["counts"]["update"] == 0
+    assert snapshot(session) == before
+
+
+def test_postgres_crawler_preserves_import_and_adopts_hse_id(pg_crawl_db, tmp_path):
+    session, _ = pg_crawl_db
+    person = employee(session)
+    cli(tmp_path, pg_crawl_db, [article(), article(title="CSV only", identifier="10.1234/only")], apply=True)
+    session.expire_all()
+    ids = {p.title: p.id for p in session.query(EmployeePublication).all()}
+    run = CrawlRun(source_url="https://synthetic.invalid/persons", status="running")
+    session.add(run)
+    session.commit()
+    parsed = {"sections": [{"type": "publications", "publications": [
+        {"publication_id": "hse-one", "title": "Synthetic article", "year": 2024,
+         "doi_url": "https://doi.org/10.1234/test", "raw_data": {"id": "hse-one"}}]}]}
+    for _ in range(2):
+        _try_sync_employee_publications(session, run, person, copy.deepcopy(parsed))
+        session.commit()
+        session.expire_all()
+    assert session.query(EmployeePublication).count() == 2
+    kept = session.get(EmployeePublication, ids["Synthetic article"])
+    assert kept.publication_id == "hse-one" and "csv_enrichment" in kept.raw_data
+    assert session.get(EmployeePublication, ids["CSV only"]).raw_data["csv_enrichment"]
+    assert session.query(CrawlError).count() == 0
+
+
+@pytest.mark.parametrize("bound_id", [None, "hse-A"])
+@pytest.mark.parametrize("incoming_ids", [("hse-A", "hse-B"), ("hse-B", "hse-A")])
+def test_postgres_hse_id_conflict_does_not_block_other_employee(pg_crawl_db, tmp_path, monkeypatch, bound_id, incoming_ids):
+    session, _ = pg_crawl_db
+    first = employee(session)
+    second = employee(session, "Synthetic Two", "two")
+    cli(tmp_path, pg_crawl_db, [article(identifier="")], apply=True)
+    session.expire_all()
+    original = session.query(EmployeePublication).filter_by(employee_id=first.id).one()
+    original.publication_id = bound_id
+    original.authors = [{"id": "original-author"}]
+    session.commit()
+    fields = ("publication_id", "source_hash", "raw_data", "authors", "updated_at")
+    before = {field: copy.deepcopy(getattr(original, field)) for field in fields}
+    session.commit()
+
+    def profile(person, publications):
+        return {"source_url": person.canonical_url, "profile_type": "staff",
+                "profile_id": person.profile_key.split(":")[1], "full_name": person.full_name,
+                "parser_version": "integration-test", "tabs": [],
+                "sections": [{"type": "publications", "publications": publications}]}
+
+    profiles = {
+        first.canonical_url: profile(first, [{"publication_id": key, "title": "Synthetic article", "year": 2024,
+                                             "authors": [{"id": key}], "raw_data": {"id": key}} for key in incoming_ids]),
+        second.canonical_url: profile(second, [{"publication_id": "second-article", "title": "Other employee article",
+                                               "year": 2024, "raw_data": {"id": "second-article"}}]),
+    }
+    monkeypatch.setattr("app.services.crawler.collect_profile_links", lambda *a, **k: list(profiles))
+    monkeypatch.setattr("app.services.crawler.parse_person_profile", lambda _, url, *a, **k: copy.deepcopy(profiles[url]))
+    settings = Settings(_env_file=None, database_url=ADMIN_URL.replace("miem_import_test", "miem_crawl_test"),
+                        source_url="https://synthetic.invalid/persons", request_delay_seconds=0)
+    run = run_crawl(session, settings)
+    session.expire_all()
+    assert run.status == "completed" and run.parsed_count == 2
+    kept = session.query(EmployeePublication).filter_by(employee_id=first.id).one()
+    assert {field: getattr(kept, field) for field in fields} == before
+    assert session.query(EmployeePublication).filter_by(employee_id=second.id).one().publication_id == "second-article"
+    error = session.query(CrawlError).filter_by(crawl_run_id=run.id).one()
+    assert "Ambiguous HSE IDs" in error.message