13 lines
397 B
Python
13 lines
397 B
Python
from kab_ingestion.orchestrator import Orchestrator
|
|
class C:
|
|
def incremental(self,t,c): return type('B',(),{'next_cursor':None})()
|
|
def full(self,t): return self.incremental(t,None)
|
|
class Cur:
|
|
def load(self,*a): return None
|
|
def save(self,*a): pass
|
|
class Pub:
|
|
def publish(self,b): pass
|
|
def test_run_success():
|
|
r=Orchestrator({'x':C()},Cur(),Pub()).run('x','t')
|
|
assert r.status == 'succeeded'
|