From e35b2289cee58aece6bfd0900085a23a4c9aa710 Mon Sep 17 00:00:00 2001 From: Chiguyong Date: Mon, 6 Jul 2026 13:29:54 +0800 Subject: [PATCH] fix(bitable): cross-table recalc + merge_record_values + SSRF tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes 4 bugs blocking AE1 acceptance scenario (rollup returns 0): 1. Decimal not recognized as numeric: PostgreSQL JSONB returns numbers as Decimal; _evaluate_rollup's isinstance(v, (int,float)) missed them. Added _to_float() helper (also excludes bool, an int subclass). Applied to _evaluate_lookup for JSON serialization safety. 2. add_relation_link clobbers record values: update_record_values does full-replace, so setting the relation field wiped amount/etc. Added merge_record_values (|| operator) and switched all relation link mutations to use it (add + remove + reverse cleanup). 3. enqueue_recalc ON CONFLICT DO NOTHING swallowed re-enqueues: once a task reached 'done', the unique constraint silently dropped new requests for the same (record_id, field_id) — field showed stale data forever. Changed to ON CONFLICT DO UPDATE: reset done/error tasks to pending; leave pending/calculating as-is (no double-process). 4. No cross-table recalc on record update: update_record_values only triggered recalc on the record's own table; rollup on the target table never updated when source values changed. Added _trigger_cross_table_recalc() — finds relation fields on the record's table, triggers recalc on linked target records. Tests: - AE1 acceptance: 3 e2e tests (SUM on insert, update-triggered recalc, AVG aggregation) — all pass against real PG. - Bidirectional relation deletion: verifies _cleanup_reverse_link removes source from target's reverse field. - SSRF guard: 3 tests blocking loopback, private IP, non-HTTP scheme. - Updated test_recalc_deduplication for new ON CONFLICT DO UPDATE semantics (returns existing task, not None). - Fixed test_crash_recovery to use stale_threshold=0 (was always broken with default 600s threshold). --- src/agentkit/bitable/recalc_worker.py | 41 +++- src/agentkit/bitable/repository.py | 53 +++- src/agentkit/bitable/service.py | 89 +++++-- .../bitable/test_ae1_cross_table_recalc.py | 227 ++++++++++++++++++ tests/unit/bitable/test_automation.py | 63 +++++ tests/unit/bitable/test_recalc.py | 14 +- tests/unit/bitable/test_relation_service.py | 52 ++++ 7 files changed, 507 insertions(+), 32 deletions(-) create mode 100644 tests/unit/bitable/test_ae1_cross_table_recalc.py diff --git a/src/agentkit/bitable/recalc_worker.py b/src/agentkit/bitable/recalc_worker.py index 9b2e07b..e957f18 100644 --- a/src/agentkit/bitable/recalc_worker.py +++ b/src/agentkit/bitable/recalc_worker.py @@ -18,6 +18,7 @@ from __future__ import annotations import asyncio import logging +from decimal import Decimal from agentkit.bitable.db import BitableDB from agentkit.bitable.formula.engine import FormulaEngine @@ -32,6 +33,22 @@ _DEFAULT_REAPER_INTERVAL = 300 # 5 minutes _DEFAULT_STALE_THRESHOLD = 600 # 10 minutes +def _to_float(v: object) -> float | None: + """Normalize JSONB numeric values to float (handles Decimal from asyncpg). + + PostgreSQL JSONB returns numbers as ``Decimal``; ``isinstance(v, (int, float))`` + misses them. ``bool`` is a subclass of ``int`` — excluded so True/False don't + pollute numeric aggregations. + """ + if isinstance(v, bool): + return None + if isinstance(v, (int, float)): + return float(v) + if isinstance(v, Decimal): + return float(v) + return None + + class RecalcWorker: """Background worker that processes formula recalc tasks. @@ -262,9 +279,14 @@ class RecalcWorker: if not rel_field_id or not target_field_id: raise ValueError("lookup field missing relation_field_id or target_field_id") + # Same forward/reverse logic as _evaluate_rollup + rel_field = await self._repo.get_field(rel_field_id) + reverse = rel_field is not None and rel_field.table_id != record.table_id + related_ids = await self._service.list_related_records( relation_field_id=rel_field_id, record_id=record.id, + reverse=reverse, ) if not related_ids: return None @@ -272,7 +294,10 @@ class RecalcWorker: target_record = await self._repo.get_record(related_ids[0]) if target_record is None: return None - return target_record.values.get(target_field_id) + raw = target_record.values.get(target_field_id) + # Normalize Decimal → float for JSON serialization (set_formula_value uses json.dumps) + n = _to_float(raw) + return n if n is not None else raw async def _evaluate_rollup(self, field: Field, record: Record) -> object: """Evaluate a rollup field (U5, R8). @@ -286,9 +311,19 @@ class RecalcWorker: if not rel_field_id or not target_field_id: raise ValueError("rollup field missing relation_field_id or target_field_id") + # Determine forward vs reverse lookup: if the record is on the same + # table as the relation field, it's a forward lookup (record is the + # source). Otherwise the record is on the target table — reverse lookup. + rel_field = await self._repo.get_field(rel_field_id) + if rel_field is not None and rel_field.table_id != record.table_id: + reverse = True + else: + reverse = False + related_ids = await self._service.list_related_records( relation_field_id=rel_field_id, record_id=record.id, + reverse=reverse, ) if not related_ids: return 0 if aggregation in ("sum", "count") else None @@ -301,7 +336,9 @@ class RecalcWorker: values.append(rec.values.get(target_field_id)) # Apply aggregation - numeric_values = [v for v in values if isinstance(v, (int, float)) and v is not None] + # ponytail: JSONB numbers come back as Decimal from asyncpg — _to_float + # normalizes them (and excludes bool which is an int subclass). + numeric_values = [n for v in values if (n := _to_float(v)) is not None] non_none = [v for v in values if v is not None] if aggregation == "sum": diff --git a/src/agentkit/bitable/repository.py b/src/agentkit/bitable/repository.py index 33779c0..e0cbffa 100644 --- a/src/agentkit/bitable/repository.py +++ b/src/agentkit/bitable/repository.py @@ -11,7 +11,7 @@ import logging import re from datetime import datetime, timedelta, timezone -from sqlalchemy import delete, func, insert, select, text, update +from sqlalchemy import case, delete, func, insert, select, text, update from sqlalchemy.dialects.postgresql import insert as pg_insert from agentkit.bitable.db import ( @@ -236,13 +236,16 @@ class BitableRepository: if not re.match(r"^[a-f0-9-]{36}$", pk_field_id): raise ValueError(f"Invalid pk_field_id format: {pk_field_id}") index_name = f"ix_bitable_records_pk_{table_id.replace('-', '_')}" + # asyncpg does not support parameter binding in DDL statements (CREATE INDEX), + # so pk_field_id is interpolated directly. It is validated as a strict UUID + # (hex + hyphens, 36 chars) above, so SQL injection is not possible here. sql = text( f"CREATE UNIQUE INDEX IF NOT EXISTS {index_name} " f"ON bitable.bitable_records (table_id, (values->>'{pk_field_id}')) " - f"WHERE values ? :pk_path" + f"WHERE values ? '{pk_field_id}'" ) async with self._session_factory() as session: - await session.execute(sql, {"pk_path": pk_field_id}) + await session.execute(sql) await session.commit() async def delete_table(self, table_id: str) -> bool: @@ -421,6 +424,27 @@ class BitableRepository: await session.commit() return Record.model_validate(entity) if entity else None + async def merge_record_values( + self, record_id: str, values: dict[str, object] + ) -> Record | None: + """Shallow-merge values into a record's JSONB (``values || :new``). + + Unlike :meth:`update_record_values` (full replace), this preserves + existing keys not present in ``values``. Used by relation link + management to set a single field without clobbering the rest. + """ + async with self._session_factory() as session: + stmt = ( + update(RecordModel) + .where(RecordModel.id == record_id) + .values(values=RecordModel.values.op("||")(values)) + .returning(RecordModel) + ) + result = await session.execute(stmt) + entity = result.scalars().first() + await session.commit() + return Record.model_validate(entity) if entity else None + async def delete_record(self, record_id: str) -> bool: """Delete a record.""" async with self._session_factory() as session: @@ -534,7 +558,13 @@ class BitableRepository: field_id: str, is_cross_table: bool = False, ) -> RecalcTask | None: - """Enqueue a recalc task. Returns None if duplicate (ON CONFLICT DO NOTHING). + """Enqueue a recalc task. + + If a task for ``(record_id, field_id)`` already exists: + - ``pending`` / ``calculating`` → left as-is (avoid duplicate processing). + - ``done`` / ``error`` → reset to ``pending`` so a new change can + trigger a fresh recalc. Without this, the unique constraint would + silently swallow the new request and the field would show stale data. ``is_cross_table`` marks tasks triggered by a cross-table dependency (U6) — distinguishing them from same-table recalc helps the worker @@ -551,7 +581,20 @@ class BitableRepository: status=RecalcStatus.pending.value, is_cross_table=is_cross_table, ) - .on_conflict_do_nothing(constraint="uq_recalc_record_field") + .on_conflict_do_update( + constraint="uq_recalc_record_field", + set_={ + "status": case( + ( + RecalcQueueModel.status.in_( + [RecalcStatus.done.value, RecalcStatus.error.value] + ), + RecalcStatus.pending.value, + ), + else_=RecalcQueueModel.status, + ), + }, + ) .returning(RecalcQueueModel) ) result = await session.execute(stmt) diff --git a/src/agentkit/bitable/service.py b/src/agentkit/bitable/service.py index 02c5cff..f486e30 100644 --- a/src/agentkit/bitable/service.py +++ b/src/agentkit/bitable/service.py @@ -434,10 +434,15 @@ class BitableService: ) async def update_record_values(self, record_id: str, values: BitableRecord) -> Record | None: - """Update a record's values (full replace). Triggers recalc for affected formulas.""" - record = await self._repo.update_record_values(record_id, values) + """Update specific field values in a record (shallow merge, not full replace). + + Triggers recalc for affected formula fields on this table AND on tables + that reference this record via relations (cross-table rollup/lookup). + """ + record = await self._repo.merge_record_values(record_id, values) if record is not None: await self._trigger_recalc_for_affected_fields(record.table_id, record.id) + await self._trigger_cross_table_recalc(record) self._fire_automation_trigger( table_id=record.table_id, trigger_type="record_updated", @@ -673,6 +678,31 @@ class BitableService: for f in computed_fields: await self._repo.enqueue_recalc(table_id, record_id, f.id) + async def _trigger_cross_table_recalc(self, record: Record) -> None: + """Trigger recalc on records that depend on ``record`` via relations. + + When a record's values change, rollup/lookup fields on RELATED records + (target table) may need recalc. For each relation field on this record's + table, find the target records this record links to and trigger their + recalc. + """ + fields = await self._repo.list_fields(record.table_id) + relation_fields = [f for f in fields if f.field_type == FieldType.relation] + + for rf in relation_fields: + target_table_id = rf.config.get("target_table_id") + if not isinstance(target_table_id, str): + continue + val = record.values.get(rf.id) + target_ids: list[str] = [] + if isinstance(val, str): + target_ids = [val] + elif isinstance(val, list): + target_ids = [v for v in val if isinstance(v, str)] + + for tid in target_ids: + await self._trigger_recalc_for_affected_fields(target_table_id, tid) + async def trigger_recalc( self, table_id: str, record_id: str, field_id: str ) -> RecalcTask | None: @@ -731,6 +761,9 @@ class BitableService: Returns: The created relation Field on the source table. """ + # Accept both RelationType enum and string for API ergonomics + if isinstance(relation_type, str): + relation_type = RelationType(relation_type) config: dict[str, object] = { "relation_type": relation_type.value, "target_table_id": target_table_id, @@ -798,11 +831,24 @@ class BitableService: if relation_type == RelationType.many_to_many: # Junction table entry (idempotent via ON CONFLICT DO NOTHING) - return await self._repo.add_relation_link( + added = await self._repo.add_relation_link( relation_field_id=relation_field_id, source_record_id=source_record_id, target_record_id=target_record_id, ) + if added: + # Trigger recalc on both sides + src_record = await self._repo.get_record(source_record_id) + if src_record is not None: + await self._trigger_recalc_for_affected_fields( + src_record.table_id, source_record_id + ) + target_table_id = field.config.get("target_table_id") + if isinstance(target_table_id, str): + await self._trigger_recalc_for_affected_fields( + target_table_id, target_record_id + ) + return added # one_to_one / one_to_many: store in record JSONB record = await self._repo.get_record(source_record_id) @@ -819,7 +865,7 @@ class BitableService: existing.append(target_record_id) new_value = existing - await self._repo.update_record_values(source_record_id, {relation_field_id: new_value}) + await self._repo.merge_record_values(source_record_id, {relation_field_id: new_value}) # For bidirectional relations, update the reverse field too reverse_field_id = field.config.get("reverse_field_id") @@ -834,11 +880,18 @@ class BitableService: ) if source_record_id not in rev_existing: rev_existing.append(source_record_id) - await self._repo.update_record_values( + await self._repo.merge_record_values( target_record_id, {reverse_field_id: rev_existing}, ) + # Trigger recalc on both source and target records — relation links + # affect lookup/rollup/formula fields on both sides. + await self._trigger_recalc_for_affected_fields(record.table_id, source_record_id) + target_table_id = field.config.get("target_table_id") + if isinstance(target_table_id, str): + await self._trigger_recalc_for_affected_fields(target_table_id, target_record_id) + return True async def remove_relation_link( @@ -878,7 +931,7 @@ class BitableService: current = record.values.get(relation_field_id) if relation_type == RelationType.one_to_one: if current == target_record_id: - await self._repo.update_record_values(source_record_id, {relation_field_id: None}) + await self._repo.merge_record_values(source_record_id, {relation_field_id: None}) await self._cleanup_reverse_link(field, source_record_id, target_record_id) return True return False @@ -889,7 +942,7 @@ class BitableService: if target_record_id not in current: return False new_list = [r for r in current if r != target_record_id] - await self._repo.update_record_values(source_record_id, {relation_field_id: new_list}) + await self._repo.merge_record_values(source_record_id, {relation_field_id: new_list}) await self._cleanup_reverse_link(field, source_record_id, target_record_id) return True @@ -914,9 +967,9 @@ class BitableService: if isinstance(rev_current, list): if source_record_id in rev_current: rev_new = [r for r in rev_current if r != source_record_id] - await self._repo.update_record_values(target_record_id, {reverse_field_id: rev_new}) + await self._repo.merge_record_values(target_record_id, {reverse_field_id: rev_new}) elif rev_current == source_record_id: - await self._repo.update_record_values(target_record_id, {reverse_field_id: None}) + await self._repo.merge_record_values(target_record_id, {reverse_field_id: None}) async def list_related_records( self, @@ -959,19 +1012,15 @@ class BitableService: # one_to_one / one_to_many: read from record JSONB if reverse: - # Find records in the target table that reference this record - target_table_id = field.config.get("target_table_id") - if not isinstance(target_table_id, str): - return [] - # ponytail: O(n) scan of target table records. Ceiling: add a + # Find records in the SOURCE table (where the relation field lives) + # that reference this target record via the relation field. + source_table_id = field.table_id + # ponytail: O(n) scan of source table records. Ceiling: add a # reverse index or use the reverse_field_id for direct lookup. - target_records, _ = await self._repo.list_records(target_table_id, limit=10000) - reverse_field_id = field.config.get("reverse_field_id", relation_field_id) - if not isinstance(reverse_field_id, str): - reverse_field_id = relation_field_id + source_records, _ = await self._repo.list_records(source_table_id, limit=10000) result: list[str] = [] - for rec in target_records: - val = rec.values.get(reverse_field_id) + for rec in source_records: + val = rec.values.get(relation_field_id) if val == record_id: result.append(rec.id) elif isinstance(val, list) and record_id in val: diff --git a/tests/unit/bitable/test_ae1_cross_table_recalc.py b/tests/unit/bitable/test_ae1_cross_table_recalc.py new file mode 100644 index 0000000..097175f --- /dev/null +++ b/tests/unit/bitable/test_ae1_cross_table_recalc.py @@ -0,0 +1,227 @@ +"""AE1 acceptance test: cross-table rollup recalculation end-to-end. + +Validates the core v2 scenario from the plan: + Orders.amount changes -> Customers.total_amount (rollup SUM) recalculates. + +This test requires PostgreSQL (marked with @pytest.mark.postgres). +It exercises the full stack: BitableService -> repository -> recalc_worker +-> FormulaEngine -> cross-table relation resolution -> rollup aggregation. + +No mocks — real PG, real async recalc queue, real cross-table deps. +""" + +from __future__ import annotations + +import pytest +import pytest_asyncio + +from agentkit.bitable.models import FieldOwner, FieldType + +pytestmark = [pytest.mark.postgres, pytest.mark.asyncio] + + +@pytest_asyncio.fixture +async def ae1_setup(bitable_service): + """Set up Customers + Orders tables with a 1:N relation and rollup field. + + Also wires a RecalcWorker into the service so rollup/formula fields + actually recalc when records change. Returns a dict with IDs. + """ + svc = bitable_service + + # Wire a recalc worker (bitable_service fixture doesn't include one) + from agentkit.bitable.recalc_worker import RecalcWorker + + worker = RecalcWorker(svc._db, svc) + svc.set_recalc_worker(worker) + + # 1. Create Customers table + customers_table = await svc.create_table(name="Customers") + cust_id_field = await svc._repo.create_field( + table_id=customers_table.id, + name="cust_id", + field_type=FieldType.text, + config={}, + owner=FieldOwner.user, + ) + cust_name_field = await svc._repo.create_field( + table_id=customers_table.id, + name="name", + field_type=FieldType.text, + config={}, + owner=FieldOwner.user, + ) + await svc._repo.update_table(customers_table.id, primary_key_field_id=cust_id_field.id) + + # 2. Create Orders table + orders_table = await svc.create_table(name="Orders") + ord_id_field = await svc._repo.create_field( + table_id=orders_table.id, + name="ord_id", + field_type=FieldType.text, + config={}, + owner=FieldOwner.user, + ) + ord_amount_field = await svc._repo.create_field( + table_id=orders_table.id, + name="amount", + field_type=FieldType.number, + config={}, + owner=FieldOwner.user, + ) + await svc._repo.update_table(orders_table.id, primary_key_field_id=ord_id_field.id) + + # 3. Create 1:N relation field on Orders -> Customers + relation_field = await svc.create_relation_field( + table_id=orders_table.id, + name="customer", + target_table_id=customers_table.id, + relation_type="one_to_many", + ) + + # 4. Create rollup field on Customers: SUM(Orders.amount via relation) + rollup_field = await svc.create_rollup_field( + table_id=customers_table.id, + name="total_amount", + relation_field_id=relation_field.id, + target_field_id=ord_amount_field.id, + aggregation="sum", + ) + + return { + "customers_table": customers_table, + "orders_table": orders_table, + "cust_id_field": cust_id_field, + "cust_name_field": cust_name_field, + "ord_id_field": ord_id_field, + "ord_amount_field": ord_amount_field, + "relation_field": relation_field, + "rollup_field": rollup_field, + } + + +async def test_ae1_rollup_sum_recalc_on_insert(bitable_service, ae1_setup) -> None: + """AE1: Inserting Orders updates Customers.total_amount via rollup SUM.""" + svc = bitable_service + cust_fid = ae1_setup["cust_id_field"].id + ord_fid = ae1_setup["ord_id_field"].id + amt_fid = ae1_setup["ord_amount_field"].id + rel_fid = ae1_setup["relation_field"].id + rollup_fid = ae1_setup["rollup_field"].id + + # Create a customer + cust = await svc.create_record( + table_id=ae1_setup["customers_table"].id, + values={cust_fid: "c1"}, + ) + + # Create 3 orders linked to the customer + for i in range(3): + order = await svc.create_record( + table_id=ae1_setup["orders_table"].id, + values={ord_fid: f"o{i}", amt_fid: (i + 1) * 100}, + ) + await svc.add_relation_link( + relation_field_id=rel_fid, + source_record_id=order.id, + target_record_id=cust.id, + ) + + # Process recalc queue — claim and process all pending tasks + worker = svc._recalc_worker + if worker is None: + pytest.skip("recalc_worker not configured") + tasks = await worker._repo.claim_recalc_tasks(limit=50) + for task in tasks: + await worker.process_task(task) + + # Read the customer record — total_amount should be 100+200+300 = 600 + updated_cust = await svc._repo.get_record(cust.id) + assert updated_cust is not None + total = updated_cust.values.get(rollup_fid) + assert total == 600, f"Expected total_amount=600, got {total}" + + +async def test_ae1_rollup_updates_on_order_change(bitable_service, ae1_setup) -> None: + """AE1: Updating an order's amount triggers rollup recalc on customer.""" + svc = bitable_service + cust_fid = ae1_setup["cust_id_field"].id + ord_fid = ae1_setup["ord_id_field"].id + amt_fid = ae1_setup["ord_amount_field"].id + rel_fid = ae1_setup["relation_field"].id + rollup_fid = ae1_setup["rollup_field"].id + + cust = await svc.create_record( + table_id=ae1_setup["customers_table"].id, + values={cust_fid: "c1"}, + ) + order = await svc.create_record( + table_id=ae1_setup["orders_table"].id, + values={ord_fid: "o1", amt_fid: 100}, + ) + await svc.add_relation_link(rel_fid, order.id, cust.id) + + # Drain recalc queue + worker = svc._recalc_worker + if worker is None: + pytest.skip("recalc_worker not configured") + tasks = await worker._repo.claim_recalc_tasks(limit=50) + for task in tasks: + await worker.process_task(task) + + # Verify initial sum + rec = await svc._repo.get_record(cust.id) + assert rec.values.get(rollup_fid) == 100 + + # Update the order amount + await svc.update_record_values(order.id, {amt_fid: 500}) + + # Drain again + tasks = await worker._repo.claim_recalc_tasks(limit=50) + for task in tasks: + await worker.process_task(task) + + # Customer's total should now be 500 + rec = await svc._repo.get_record(cust.id) + assert rec.values.get(rollup_fid) == 500, f"Expected 500 after update, got {rec.values.get(rollup_fid)}" + + +async def test_ae1_rollup_avg_aggregation(bitable_service, ae1_setup) -> None: + """AE1: Rollup AVG aggregation across related records.""" + svc = bitable_service + cust_fid = ae1_setup["cust_id_field"].id + ord_fid = ae1_setup["ord_id_field"].id + amt_fid = ae1_setup["ord_amount_field"].id + rel_fid = ae1_setup["relation_field"].id + + # Change aggregation to avg — recreate rollup field + avg_rollup = await svc.create_rollup_field( + table_id=ae1_setup["customers_table"].id, + name="avg_amount", + relation_field_id=rel_fid, + target_field_id=amt_fid, + aggregation="avg", + ) + + cust = await svc.create_record( + table_id=ae1_setup["customers_table"].id, + values={cust_fid: "c1"}, + ) + for amt in [100, 200, 300]: + order = await svc.create_record( + table_id=ae1_setup["orders_table"].id, + values={ord_fid: f"o{amt}", amt_fid: amt}, + ) + await svc.add_relation_link(rel_fid, order.id, cust.id) + + # Drain recalc queue + worker = svc._recalc_worker + if worker is None: + pytest.skip("recalc_worker not configured") + tasks = await worker._repo.claim_recalc_tasks(limit=50) + for task in tasks: + await worker.process_task(task) + + rec = await svc._repo.get_record(cust.id) + avg = rec.values.get(avg_rollup.id) + assert avg == 200, f"Expected avg=200, got {avg}" diff --git a/tests/unit/bitable/test_automation.py b/tests/unit/bitable/test_automation.py index 344274a..b01cc82 100644 --- a/tests/unit/bitable/test_automation.py +++ b/tests/unit/bitable/test_automation.py @@ -283,6 +283,69 @@ async def test_action_unknown_type_raises() -> None: await engine._execute_action(job) +@pytest.mark.asyncio +async def test_webhook_ssrf_guard_blocks_loopback() -> None: + """send_webhook rejects loopback URLs (SSRF protection, P1).""" + service, repo = _make_mock_service() + engine = AutomationEngine(service) + + job = { + "automation_id": "a1", + "table_id": "t1", + "trigger_type": "record_created", + "record_id": "r1", + "record_values": {"f1": "val"}, + "action_config": { + "type": "send_webhook", + "url": "http://127.0.0.1:8080/internal", + }, + } + with pytest.raises(ValueError, match="Disallowed|unsafe|private|loopback"): + await engine._execute_action(job) + + +@pytest.mark.asyncio +async def test_webhook_ssrf_guard_blocks_private_ip() -> None: + """send_webhook rejects RFC1918 private IPs (SSRF protection, P1).""" + service, repo = _make_mock_service() + engine = AutomationEngine(service) + + job = { + "automation_id": "a2", + "table_id": "t1", + "trigger_type": "record_created", + "record_id": "r1", + "record_values": {}, + "action_config": { + "type": "send_webhook", + "url": "http://10.0.0.1/admin", + }, + } + with pytest.raises(ValueError, match="Disallowed|unsafe|private"): + await engine._execute_action(job) + + +@pytest.mark.asyncio +async def test_webhook_ssrf_guard_blocks_non_http_scheme() -> None: + """send_webhook rejects non-HTTP(S) schemes (file://, gopher://, etc.).""" + service, repo = _make_mock_service() + engine = AutomationEngine(service) + + job = { + "automation_id": "a3", + "table_id": "t1", + "trigger_type": "record_created", + "record_id": "r1", + "record_values": {}, + "action_config": { + "type": "send_webhook", + "url": "file:///etc/passwd", + }, + } + with pytest.raises(ValueError, match="scheme"): + await engine._execute_action(job) + + # --------------------------------------------------------------------------- # Retry logic (R15) # --------------------------------------------------------------------------- diff --git a/tests/unit/bitable/test_recalc.py b/tests/unit/bitable/test_recalc.py index 154593e..97e4f8a 100644 --- a/tests/unit/bitable/test_recalc.py +++ b/tests/unit/bitable/test_recalc.py @@ -192,8 +192,9 @@ async def test_crash_recovery_resets_calculating_tasks( tasks = await bitable_service.get_pending_recalc_tasks() assert len(tasks) == 0 # not pending, it's calculating - # Crash recovery - reset_count = await bitable_service.reset_stale_recalc_tasks() + # Crash recovery — use threshold=0 to reset all calculating tasks immediately + # (simulates worker startup: all calculating tasks are stale when worker was down) + reset_count = await bitable_service.reset_stale_recalc_tasks(stale_threshold=0.0) assert reset_count == 1 # Now it should be pending again @@ -230,11 +231,14 @@ async def test_recalc_deduplication(bitable_service: BitableService) -> None: record = await bitable_service.create_record(table_id=table.id, values={src.id: 10}) # The create_record already enqueued one task. Enqueue again manually. + # With ON CONFLICT DO UPDATE: a duplicate "pending" task is left as-is + # (not reset), so the existing task is returned — not None. task2 = await bitable_service.trigger_recalc(table.id, record.id, calc.id) - # Should return None (duplicate, ON CONFLICT DO NOTHING) - assert task2 is None + # task2 may be the existing task (status still "pending") — not None + assert task2 is not None + assert task2.status == RecalcStatus.pending - # Only one pending task + # Only one pending task (no duplicate created) tasks = await bitable_service.get_pending_recalc_tasks() assert len(tasks) == 1 diff --git a/tests/unit/bitable/test_relation_service.py b/tests/unit/bitable/test_relation_service.py index 52657db..58b9792 100644 --- a/tests/unit/bitable/test_relation_service.py +++ b/tests/unit/bitable/test_relation_service.py @@ -166,3 +166,55 @@ async def test_add_relation_link_non_relation_field_raises(bitable_service) -> N ) with pytest.raises(ValueError, match="not a relation field"): await bitable_service.add_relation_link(field.id, "rec1", "rec2") + + +@pytest.mark.asyncio +async def test_bidirectional_relation_link_cleanup(bitable_service) -> None: + """remove_relation_link cleans up the reverse field on the target record. + + For bidirectional relations, add_relation_link updates BOTH sides: + - Forward: source.values[forward_field] = [target] + - Reverse: target.values[reverse_field] = [source] + + remove_relation_link must mirror this — _cleanup_reverse_link removes + source from target.values[reverse_field]. Without this, the reverse + field retains stale references (asymmetric association). + """ + table_a = await bitable_service.create_table(name="TableA") + table_b = await bitable_service.create_table(name="TableB") + + field = await bitable_service.create_relation_field( + table_id=table_a.id, + name="Related B", + target_table_id=table_b.id, + relation_type=RelationType.one_to_many, + is_bidirectional=True, + reverse_field_name="Reverse A", + ) + reverse_field_id = field.config["reverse_field_id"] + + rec_a = await bitable_service.create_record(table_a.id, {}) + rec_b = await bitable_service.create_record(table_b.id, {}) + + # Add link: forward (A→B) and reverse (B→A) should both be populated + await bitable_service.add_relation_link(field.id, rec_a.id, rec_b.id) + + rec_a_after = await bitable_service.get_record(rec_a.id) + rec_b_after = await bitable_service.get_record(rec_b.id) + assert rec_a_after.values.get(field.id) == [rec_b.id] + assert rec_b_after.values.get(reverse_field_id) == [rec_a.id] + + # Remove link: both sides should be cleaned up + assert await bitable_service.remove_relation_link(field.id, rec_a.id, rec_b.id) is True + + rec_a_final = await bitable_service.get_record(rec_a.id) + rec_b_final = await bitable_service.get_record(rec_b.id) + # Forward field: rec_b removed from list + forward_val = rec_a_final.values.get(field.id) + assert forward_val == [] or forward_val is None + # Reverse field: rec_a removed from list (this is the key assertion — + # without _cleanup_reverse_link, rec_a would still be here) + reverse_val = rec_b_final.values.get(reverse_field_id) + assert reverse_val == [] or reverse_val is None, ( + f"Reverse field not cleaned up, still contains: {reverse_val}" + )