digdag for testing.

fake_digdag.py

import sys
import imp
import collections

digdag_env_mod = sys.modules['digdag_env'] = imp.new_module('digdag_env')

digdag_env_mod.params = {}
digdag_env_mod.subtask_config = collections.OrderedDict()
digdag_env_mod.export_params = {}
digdag_env_mod.store_params = {}
digdag_env_mod.state_params = {}

digdag_mod = sys.modules['digdag'] = imp.new_module('digdag')

class Env(object):
    def __init__(self, digdag_env_mod):
        self.params = digdag_env_mod.params
        self.subtask_config = digdag_env_mod.subtask_config
        self.export_params = digdag_env_mod.export_params
        self.store_params = digdag_env_mod.store_params
        self.state_params = digdag_env_mod.state_params
        self.subtask_index = 0

    def set_state(self, params={}, **kwds):
        self.state_params.update(params)
        self.state_params.update(kwds)

    def export(self, params={}, **kwds):
        self.export_params.update(params)
        self.export_params.update(kwds)

    def store(self, params={}, **kwds):
        self.store_params.update(params)
        self.store_params.update(kwds)

digdag_mod.env = Env(digdag_env_mod)