17 lines
756 B
Python
17 lines
756 B
Python
from kab_ingestion.orchestrator import Orchestrator
|
|
from kab_ingestion.models import *
|
|
class C:
|
|
def full(self,c): return [],SyncCursor(c.connector_id,"full")
|
|
def incremental(self,c,cur,changes=None): return [],cur
|
|
def webhook(self,c,p,h): return []
|
|
class P:
|
|
def upsert(self,d): self.d=list(d)
|
|
def delete(self,i,t): pass
|
|
def test_run_is_idempotent_shape():
|
|
p=P(); r=Orchestrator({"x":C()},p).run(ConnectorConfig("x","t","s",{}),mode="full",run_id="fixed")
|
|
assert r.status=="succeeded" and r.run_id=="fixed"
|
|
def test_validation_rejects_cross_tenant():
|
|
from kab_ingestion.validation import validate_documents
|
|
d=Document("i","other","x","x","text/plain","h",None,Provenance("x","i","u"),ACL("other"))
|
|
assert not validate_documents([d],"t")["passed"]
|