mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-30 12:25:29 +08:00
583 lines
23 KiB
Python
583 lines
23 KiB
Python
"""FastGPT deletion safety: dispatched-but-unknown deletes, binding, and uploads.
|
|
|
|
Local HTTP fixture; real SDK objects via ``test_shared_state``'s loader.
|
|
"""
|
|
import asyncio
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
from test_shared_state import (
|
|
StorageFixture,
|
|
bind,
|
|
configuration,
|
|
ingest_context,
|
|
load_plugin,
|
|
)
|
|
|
|
from langbot_plugin.api.entities.builtin.rag import DocumentStatus
|
|
|
|
|
|
def bind_mock_http(monkeypatch, handler):
|
|
client = httpx.AsyncClient
|
|
monkeypatch.setattr(
|
|
httpx, 'AsyncClient',
|
|
lambda **kw: client(transport=httpx.MockTransport(handler), **kw),
|
|
)
|
|
|
|
|
|
def upload_engine(pc, ec, store):
|
|
"""An engine whose Host resolves storage and file streaming to local fixtures."""
|
|
engine = ec()
|
|
plugin = bind(pc, store)
|
|
|
|
async def read_file(path):
|
|
return b'fixture file'
|
|
|
|
plugin.get_knowledge_file_stream = read_file
|
|
engine.plugin = plugin
|
|
return engine
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fastgpt_cancelled_delete_keeps_fence_and_mapping(monkeypatch):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
dispatched = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def service(request):
|
|
if request.method == 'DELETE':
|
|
dispatched.set()
|
|
await release.wait()
|
|
return httpx.Response(200, json={'code': 200, 'data': None})
|
|
|
|
bind_mock_http(monkeypatch, service)
|
|
store = StorageFixture()
|
|
engine = ec()
|
|
engine.plugin = bind(pc, store)
|
|
cfg = configuration('A')
|
|
await engine.on_knowledge_base_create('kb', cfg)
|
|
await engine._save_document('kb', 'host-doc', {
|
|
'upstream_id': 'upstream-doc', 'dataset_id': 'A', 'status': 'created',
|
|
'api_base_url': cfg['api_base_url']})
|
|
task = asyncio.create_task(engine.delete_document('kb', 'host-doc'))
|
|
await dispatched.wait()
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
release.set()
|
|
# The delete was already dispatched, so its outcome is unknown: the mapping
|
|
# must not be closed out as deleted.
|
|
assert (await engine._load_document('kb', 'host-doc'))['status'] == 'created'
|
|
# The knowledge base stays fenced, also for a worker that restarts.
|
|
with pytest.raises(RuntimeError, match='fenced'):
|
|
await engine.delete_document('kb', 'host-doc')
|
|
restarted = ec()
|
|
restarted.plugin = bind(pc, store)
|
|
with pytest.raises(RuntimeError, match='fenced'):
|
|
await restarted.delete_document('kb', 'host-doc')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('recorded_target', [None, 'https://other.invalid'])
|
|
async def test_fastgpt_delete_refuses_mapping_from_another_upstream(
|
|
monkeypatch, recorded_target
|
|
):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
calls = []
|
|
|
|
async def service(request):
|
|
calls.append(request.method)
|
|
return httpx.Response(200, json={'code': 200, 'data': None})
|
|
|
|
bind_mock_http(monkeypatch, service)
|
|
store = StorageFixture()
|
|
engine = ec()
|
|
engine.plugin = bind(pc, store)
|
|
cfg = configuration('A')
|
|
await engine.on_knowledge_base_create('kb', cfg)
|
|
mapping = {'upstream_id': 'upstream-doc', 'dataset_id': 'A', 'status': 'created'}
|
|
if recorded_target is not None:
|
|
mapping['api_base_url'] = recorded_target
|
|
await engine._save_document('kb', 'host-doc', mapping)
|
|
# A mapping that is not bound to the configured upstream target (including a
|
|
# mapping recorded before targets were bound) is refused before any request.
|
|
with pytest.raises(RuntimeError, match='Upstream target changed'):
|
|
await engine.delete_document('kb', 'host-doc')
|
|
assert calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fastgpt_deterministic_upload_rejection_leaves_no_fence(monkeypatch):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
|
|
async def service(request):
|
|
return httpx.Response(400, json={'message': 'unsupported file'})
|
|
|
|
bind_mock_http(monkeypatch, service)
|
|
store = StorageFixture()
|
|
engine = upload_engine(pc, ec, store)
|
|
result = await engine.ingest(ingest_context(configuration('A')))
|
|
assert result.status == DocumentStatus.FAILED
|
|
# The provider answered, so the pre-dispatch upload intent is cleared again:
|
|
# a restarted worker accepts mutations for this knowledge base.
|
|
restarted = upload_engine(pc, ec, store)
|
|
await restarted.on_knowledge_base_create('kb', configuration('A'))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fastgpt_upload_binds_mapping_to_normalized_upstream_target(monkeypatch):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
|
|
async def service(request):
|
|
if request.method == 'POST':
|
|
return httpx.Response(200, json={
|
|
'code': 200, 'data': {'collectionId': 'upstream-doc', 'results': {}},
|
|
})
|
|
return httpx.Response(200, json={'code': 200, 'data': None})
|
|
|
|
bind_mock_http(monkeypatch, service)
|
|
store = StorageFixture()
|
|
engine = upload_engine(pc, ec, store)
|
|
cfg = configuration('A')
|
|
# Casing and a trailing slash are config spellings, not a different upstream.
|
|
upload = await engine.ingest(ingest_context({
|
|
**cfg, 'api_base_url': cfg['api_base_url'].upper() + '/',
|
|
}))
|
|
assert upload.status == DocumentStatus.PROCESSING
|
|
mapping = await engine._load_document('kb', 'local-doc')
|
|
assert mapping['api_base_url'] == cfg['api_base_url']
|
|
# The mapping names the upstream that owns the collection, so a delete
|
|
# against that same upstream still succeeds.
|
|
assert await engine.delete_document('kb', 'local-doc') is True
|
|
|
|
|
|
def fence_key_fault(shared_state, message):
|
|
def fault(key):
|
|
return RuntimeError(message) if key.startswith(shared_state.FENCE_KEY_PREFIX) else None
|
|
return fault
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fastgpt_unpersistable_fence_aborts_the_upload_before_the_request(monkeypatch):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
from components import shared_state
|
|
|
|
calls = []
|
|
|
|
async def service(request):
|
|
calls.append(request)
|
|
return httpx.Response(200, json={
|
|
'code': 200, 'data': {'collectionId': 'upstream-doc', 'results': {}},
|
|
})
|
|
|
|
bind_mock_http(monkeypatch, service)
|
|
store = StorageFixture()
|
|
store.write_fault = fence_key_fault(shared_state, 'fixture fence write rejected')
|
|
engine = upload_engine(pc, ec, store)
|
|
|
|
# The pre-written protection cannot be persisted, so the upload must not be
|
|
# dispatched at all.
|
|
with pytest.raises(shared_state.FencePersistError, match='Could not persist fence'):
|
|
await engine.ingest(ingest_context(configuration('A')))
|
|
assert calls == []
|
|
# A mutation that was never dispatched leaves no fence behind...
|
|
assert not engine._state.is_fenced(engine._state.binding(engine.plugin), 'kb')
|
|
assert [key for key in store.data if key.startswith(shared_state.FENCE_KEY_PREFIX)] == []
|
|
# ...only the durable pre-dispatch intent it had already recorded.
|
|
assert (await engine._load_document('kb', 'local-doc'))['status'] == 'pending'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fastgpt_unpersistable_fence_aborts_the_delete_before_the_request(monkeypatch):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
from components import shared_state
|
|
|
|
calls = []
|
|
|
|
async def service(request):
|
|
calls.append(request)
|
|
return httpx.Response(200, json={'code': 200, 'data': None})
|
|
|
|
bind_mock_http(monkeypatch, service)
|
|
store = StorageFixture()
|
|
engine = ec()
|
|
engine.plugin = bind(pc, store)
|
|
cfg = configuration('A')
|
|
await engine.on_knowledge_base_create('kb', cfg)
|
|
await engine._save_document('kb', 'host-doc', {
|
|
'upstream_id': 'upstream-doc', 'dataset_id': 'A', 'status': 'created',
|
|
'api_base_url': cfg['api_base_url']})
|
|
|
|
store.write_fault = fence_key_fault(shared_state, 'fixture fence write rejected')
|
|
with pytest.raises(shared_state.FencePersistError, match='Could not persist fence'):
|
|
await engine.delete_document('kb', 'host-doc')
|
|
assert calls == []
|
|
assert not engine._state.is_fenced(engine._state.binding(engine.plugin), 'kb')
|
|
assert [key for key in store.data if key.startswith(shared_state.FENCE_KEY_PREFIX)] == []
|
|
# The delete was never dispatched, so its mapping is still the recorded one.
|
|
assert (await engine._load_document('kb', 'host-doc'))['status'] == 'created'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fastgpt_cancellation_after_ack_before_the_mapping_keeps_the_fence(monkeypatch):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
|
|
calls = []
|
|
|
|
async def service(request):
|
|
calls.append(request)
|
|
return httpx.Response(200, json={'code': 200, 'data': None})
|
|
|
|
bind_mock_http(monkeypatch, service)
|
|
store = StorageFixture()
|
|
engine = ec()
|
|
engine.plugin = bind(pc, store)
|
|
cfg = configuration('A')
|
|
await engine.on_knowledge_base_create('kb', cfg)
|
|
await engine._save_document('kb', 'host-doc', {
|
|
'upstream_id': 'upstream-doc', 'dataset_id': 'A', 'status': 'created',
|
|
'api_base_url': cfg['api_base_url']})
|
|
|
|
# The documented window: the provider has acknowledged the delete, the Host
|
|
# record of it has not been written yet.
|
|
entered = asyncio.Event()
|
|
release = asyncio.Event()
|
|
record = engine._save_document
|
|
|
|
async def gated_record(kb_id, host_document_id, value):
|
|
if value.get('status') == 'deleted':
|
|
entered.set()
|
|
await release.wait()
|
|
return await record(kb_id, host_document_id, value)
|
|
|
|
monkeypatch.setattr(engine, '_save_document', gated_record)
|
|
task = asyncio.create_task(engine.delete_document('kb', 'host-doc'))
|
|
await asyncio.wait_for(entered.wait(), 5)
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
release.set()
|
|
|
|
# The delete was acknowledged upstream but its Host record does not exist, so
|
|
# the protection must not have been released: a retry would replay the delete.
|
|
assert (await engine._load_document('kb', 'host-doc'))['status'] == 'created'
|
|
assert engine._state.is_fenced(engine._state.binding(engine.plugin), 'kb')
|
|
with pytest.raises(RuntimeError, match='fenced'):
|
|
await engine.delete_document('kb', 'host-doc')
|
|
assert [request.method for request in calls] == ['DELETE']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fastgpt_later_success_never_clears_an_unresolved_fence():
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
from components import shared_state
|
|
|
|
store = StorageFixture()
|
|
engine = ec()
|
|
engine.plugin = bind(pc, store)
|
|
plugin = engine.plugin
|
|
await engine._state.fence(plugin, 'kb', 'earlier delete outcome unknown')
|
|
|
|
ran = []
|
|
|
|
async def later_mutation():
|
|
ran.append(1)
|
|
return 'dispatched'
|
|
|
|
# The later mutation is refused instead of running and releasing the earlier
|
|
# operation's fence as if its own success had resolved it.
|
|
with pytest.raises(shared_state.FencedKnowledgeBaseError, match='unresolved'):
|
|
await engine._state.dispatched(plugin, 'kb', later_mutation, 'later mutation')
|
|
assert ran == []
|
|
assert engine._state.is_fenced(engine._state.binding(plugin), 'kb')
|
|
assert [key for key in store.data if key.startswith(shared_state.FENCE_KEY_PREFIX)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('status', [500, 400])
|
|
async def test_fastgpt_delete_outcome_is_unknown_exactly_on_a_server_error(monkeypatch, status):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
from components import shared_state
|
|
|
|
calls = []
|
|
|
|
async def service(request):
|
|
calls.append(request.method)
|
|
return httpx.Response(status, json={'message': 'upstream failed'})
|
|
|
|
bind_mock_http(monkeypatch, service)
|
|
store = StorageFixture()
|
|
engine = ec()
|
|
engine.plugin = bind(pc, store)
|
|
cfg = configuration('A')
|
|
await engine.on_knowledge_base_create('kb', cfg)
|
|
await engine._save_document('kb', 'host-doc', {
|
|
'upstream_id': 'upstream-doc', 'dataset_id': 'A', 'status': 'created',
|
|
'api_base_url': cfg['api_base_url']})
|
|
|
|
if status >= 500:
|
|
# The provider or a gateway failed after receiving the delete, so whether
|
|
# it was applied is unknown: the fence stays and the caller is told
|
|
# instead of being handed a resolved rejection.
|
|
with pytest.raises(shared_state.AmbiguousMutationError):
|
|
await engine.delete_document('kb', 'host-doc')
|
|
assert engine._state.is_fenced(engine._state.binding(engine.plugin), 'kb')
|
|
# The fence is durable, not just in this process.
|
|
assert json.loads(store.data[shared_state._kb_fence_key('kb')]) is not None
|
|
# Nothing was recorded as deleted, and the next mutation for this
|
|
# knowledge base is refused, also for a worker that restarts.
|
|
assert (await engine._load_document('kb', 'host-doc'))['status'] == 'created'
|
|
with pytest.raises(RuntimeError, match='fenced'):
|
|
await engine.delete_document('kb', 'host-doc')
|
|
restarted = ec()
|
|
restarted.plugin = bind(pc, store)
|
|
with pytest.raises(RuntimeError, match='fenced'):
|
|
await restarted.delete_document('kb', 'host-doc')
|
|
else:
|
|
# The provider answered with a rejection, so the delete did not happen.
|
|
assert await engine.delete_document('kb', 'host-doc') is False
|
|
assert not engine._state.is_fenced(engine._state.binding(engine.plugin), 'kb')
|
|
assert json.loads(store.data[shared_state._kb_fence_key('kb')]) is None
|
|
restarted = ec()
|
|
restarted.plugin = bind(pc, store)
|
|
assert await restarted.delete_document('kb', 'host-doc') is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('status', [500, 400])
|
|
async def test_fastgpt_upload_outcome_is_unknown_exactly_on_a_server_error(monkeypatch, status):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
from components import shared_state
|
|
|
|
async def service(request):
|
|
return httpx.Response(status, json={'message': 'upstream failed'})
|
|
|
|
bind_mock_http(monkeypatch, service)
|
|
store = StorageFixture()
|
|
engine = upload_engine(pc, ec, store)
|
|
cfg = configuration('A')
|
|
|
|
if status >= 500:
|
|
# A dispatched upload whose answer is a server error has an unknown
|
|
# outcome, so it is neither reported as a plain failure nor unfenced.
|
|
with pytest.raises(shared_state.AmbiguousMutationError):
|
|
await engine.ingest(ingest_context(cfg))
|
|
assert engine._state.is_fenced(engine._state.binding(engine.plugin), 'kb')
|
|
assert json.loads(store.data[shared_state._kb_fence_key('kb')]) is not None
|
|
assert (await engine._load_document('kb', 'local-doc'))['status'] == 'pending'
|
|
with pytest.raises(RuntimeError, match='fenced'):
|
|
await engine.on_knowledge_base_create('kb', cfg)
|
|
restarted = upload_engine(pc, ec, store)
|
|
with pytest.raises(RuntimeError, match='fenced'):
|
|
await restarted.on_knowledge_base_create('kb', cfg)
|
|
else:
|
|
# The provider answered with a rejection, so the upload is a failure with
|
|
# a resolved outcome and the knowledge base stays usable.
|
|
result = await engine.ingest(ingest_context(cfg))
|
|
assert result.status == DocumentStatus.FAILED
|
|
assert not engine._state.is_fenced(engine._state.binding(engine.plugin), 'kb')
|
|
await engine.on_knowledge_base_create('kb', cfg)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fastgpt_concurrent_first_load_cannot_resurrect_a_cleared_fence(monkeypatch):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
from components import shared_state
|
|
|
|
store = StorageFixture()
|
|
seed = ec()
|
|
seed.plugin = bind(pc, store)
|
|
# An earlier ambiguous mutation left this knowledge base durably fenced.
|
|
await seed._state.fence(seed.plugin, 'cleared', 'earlier delete outcome unknown')
|
|
|
|
# The restarted worker's first fence load blocks on that record, so the
|
|
# delete of the fenced knowledge base is issued while the read is in flight.
|
|
store.gate_read(shared_state._kb_fence_key('cleared'))
|
|
engine = ec()
|
|
engine.plugin = bind(pc, store)
|
|
cfg = configuration('A')
|
|
loading = asyncio.create_task(engine.on_knowledge_base_create('fresh', cfg))
|
|
await asyncio.wait_for(store.gate_entered.wait(), 5)
|
|
clearing = asyncio.create_task(engine.on_knowledge_base_delete('cleared'))
|
|
for _ in range(200):
|
|
await asyncio.sleep(0)
|
|
# Both first calls share one Host read instead of one read per knowledge base.
|
|
assert store.keys_reads == 1
|
|
store.gate_release.set()
|
|
await loading
|
|
await clearing
|
|
|
|
assert store.keys_reads == 1
|
|
# The cleared knowledge base is usable: the snapshot that was read before
|
|
# the clear was not installed in its place.
|
|
assert not engine._state.is_fenced(engine._state.binding(engine.plugin), 'cleared')
|
|
await engine.on_knowledge_base_create('cleared', cfg)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fastgpt_load_discards_a_snapshot_read_before_a_sibling_clear():
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
from components import shared_state
|
|
|
|
store = StorageFixture()
|
|
seed = ec()
|
|
seed.plugin = bind(pc, store)
|
|
await seed._state.fence(seed.plugin, 'kb', 'earlier delete outcome unknown')
|
|
|
|
store.gate_read(shared_state._kb_fence_key('kb'))
|
|
engine = ec()
|
|
engine.plugin = bind(pc, store)
|
|
cfg = configuration('A')
|
|
loading = asyncio.create_task(engine.on_knowledge_base_create('kb', cfg))
|
|
await asyncio.wait_for(store.gate_entered.wait(), 5)
|
|
# The operator path out of a fence lands while the load's Host read is in
|
|
# flight and already holds the fence record in hand.
|
|
await engine._state.clear_fence(engine.plugin, 'kb')
|
|
store.gate_release.set()
|
|
await loading
|
|
|
|
# The stale snapshot was discarded and re-read instead of installed: the
|
|
# cleared knowledge base neither fails this call nor stays fenced.
|
|
assert not engine._state.is_fenced(engine._state.binding(engine.plugin), 'kb')
|
|
assert store.keys_reads == 2
|
|
await engine.on_knowledge_base_create('kb', configuration('B'))
|
|
|
|
|
|
def collection_context(config, collection_id):
|
|
"""An ingestion context whose collection id is not the knowledge-base id."""
|
|
context = ingest_context(config)
|
|
context.collection_id = collection_id
|
|
return context
|
|
|
|
|
|
def accepting_upload():
|
|
async def service(request):
|
|
if request.method == 'POST':
|
|
return httpx.Response(200, json={
|
|
'code': 200, 'data': {'collectionId': 'upstream-1', 'results': {'insertLen': 3}},
|
|
})
|
|
return httpx.Response(200, json={'code': 200, 'data': None})
|
|
return service
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fastgpt_collection_and_knowledge_base_ids_share_one_identity(monkeypatch):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
from components import shared_state
|
|
|
|
calls = []
|
|
|
|
async def service(request):
|
|
calls.append(request.method)
|
|
return await accepting_upload()(request)
|
|
|
|
bind_mock_http(monkeypatch, service)
|
|
store = StorageFixture()
|
|
engine = upload_engine(pc, ec, store)
|
|
plugin = engine.plugin
|
|
binding = engine._state.binding(plugin)
|
|
cfg = configuration('A')
|
|
|
|
# A fence taken for the knowledge base the Host names must refuse an
|
|
# ingestion whose context carries a different collection id: the two ids are
|
|
# one knowledge base, not two independently protected scopes.
|
|
await engine._state.fence(plugin, 'kb', 'earlier upload outcome unknown')
|
|
context = collection_context(cfg, 'collection-1')
|
|
with pytest.raises(shared_state.FencedKnowledgeBaseError):
|
|
await engine.ingest(context)
|
|
assert calls == []
|
|
|
|
await engine._state.clear_fence(plugin, 'kb')
|
|
entered, release = asyncio.Event(), asyncio.Event()
|
|
|
|
async def read_file(path):
|
|
entered.set()
|
|
await release.wait()
|
|
return b'fixture file'
|
|
|
|
engine.plugin.get_knowledge_file_stream = read_file
|
|
upload = asyncio.create_task(engine.ingest(context))
|
|
await asyncio.wait_for(entered.wait(), 5)
|
|
# In flight, the mutation holds the knowledge base's lock, not the
|
|
# collection's.
|
|
assert (binding, 'kb') in engine._state._locks
|
|
assert (binding, 'collection-1') not in engine._state._locks
|
|
release.set()
|
|
assert (await upload).status == DocumentStatus.PROCESSING
|
|
|
|
# One identity for storage too: the mapping is written under the key the
|
|
# deletion that names the knowledge base resolves.
|
|
assert (await engine._load_document('kb', 'local-doc'))['upstream_id'] == 'upstream-1'
|
|
assert await engine.delete_document('kb', 'local-doc') is True
|
|
assert calls == ['POST', 'DELETE']
|
|
assert (await engine._load_document('kb', 'local-doc'))['status'] == 'deleted'
|
|
assert not engine._state.is_fenced(binding, 'kb')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fastgpt_delete_under_another_knowledge_base_never_adopts_its_mapping(monkeypatch):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
|
|
calls = []
|
|
|
|
async def service(request):
|
|
calls.append(request.method)
|
|
return httpx.Response(200, json={'code': 200, 'data': None})
|
|
|
|
bind_mock_http(monkeypatch, service)
|
|
store = StorageFixture()
|
|
engine = ec()
|
|
engine.plugin = bind(pc, store)
|
|
cfg = configuration('A')
|
|
# Knowledge base A's mapping and configuration name the upstream document;
|
|
# knowledge base B has neither.
|
|
await engine._save_config('A', cfg)
|
|
await engine._save_document('A', 'shared-doc', {
|
|
'upstream_id': 'upstream-doc', 'dataset_id': 'A', 'status': 'created',
|
|
'api_base_url': cfg['api_base_url']})
|
|
|
|
# Deleting under B must not resolve, copy or tombstone A's record: the
|
|
# operation is refused with nothing dispatched and nothing written under B.
|
|
assert await engine.delete_document('B', 'shared-doc') is False
|
|
assert calls == []
|
|
assert engine._document_key('B', 'shared-doc') not in store.data
|
|
assert engine._key('B') not in store.data
|
|
assert (await engine._load_document('A', 'shared-doc'))['status'] == 'created'
|
|
|
|
# A's own delete still reaches the provider: B never consumed its record.
|
|
assert await engine.delete_document('A', 'shared-doc') is True
|
|
assert calls == ['DELETE']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fastgpt_pending_record_of_another_knowledge_base_does_not_block_ingest(monkeypatch):
|
|
pc, ec = load_plugin('FastGPTConnector')
|
|
|
|
calls = []
|
|
|
|
async def service(request):
|
|
calls.append(request.method)
|
|
return await accepting_upload()(request)
|
|
|
|
bind_mock_http(monkeypatch, service)
|
|
store = StorageFixture()
|
|
engine = upload_engine(pc, ec, store)
|
|
# A pending record under knowledge base A is A's upload intent, not B's.
|
|
await engine._save_document('A', 'local-doc', {
|
|
'upstream_id': '', 'dataset_id': 'A',
|
|
'api_base_url': 'https://fixture.invalid', 'status': 'pending'})
|
|
before = store.data[engine._document_key('A', 'local-doc')]
|
|
|
|
context = ingest_context(configuration('B'))
|
|
context.knowledge_base_id = 'B'
|
|
result = await engine.ingest(context)
|
|
assert result.status == DocumentStatus.PROCESSING
|
|
|
|
# B recorded only its own mapping, and A's pending record is untouched: B
|
|
# neither treated A's intent as its own nor wrote under A.
|
|
assert (await engine._load_document('B', 'local-doc'))['status'] == 'created'
|
|
assert (await engine._load_document('A', 'local-doc'))['status'] == 'pending'
|
|
assert store.data[engine._document_key('A', 'local-doc')] == before
|
|
assert calls == ['POST']
|