Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_add_subs(self):
class FakeSubCMD(CMD):
arg_test = Arg("-test")
class FakeCMD(CMD):
sub_cmd = FakeSubCMD
parser = Parser()
with patch.object(parser, "add_subparsers") as mock_method:
parser.add_subs(FakeCMD)
mock_method.assert_called_once()
parser = Parser()
with patch.object(parser, "add_subparsers") as mock_method:
parser.add_subs(FakeSubCMD)
with self.assertRaisesRegex(AssertionError, "Called 0 times"):
mock_method.assert_called_once()
def test_display_docstring(self):
class FakeClass(CMD):
"docstring!"
with patch.object(sys.stdout, "write") as mock_method:
ListEntrypoint().display(FakeClass)
mock_method.assert_any_call("docstring!")
async def test_cli_run_sub_command(self):
class Secondary(CMD):
async def run(self):
return 2
class Primary(CMD):
secondary = Secondary
self.assertEqual(await Primary.cli("secondary"), 2)
class Edit(SourcesCMD, KeysCMD):
"""
Edit each specified repo
"""
async def run(self):
async with self.sources as sources:
async with sources() as sctx:
for key in self.keys:
repo = await sctx.repo(key)
pdb.set_trace()
await sctx.update(repo)
class Merge(CMD):
"""
Merge repo data between sources
"""
arg_dest = Arg(
"dest", help="Sources merge repos into", type=BaseSource.load_labeled
)
arg_src = Arg(
"src", help="Sources to pull repos from", type=BaseSource.load_labeled
)
async def run(self):
async with self.src.withconfig(
self.extra_config
) as src, self.dest.withconfig(self.extra_config) as dest:
async with src() as sctx, dest() as dctx:
skel = Skel()
async def run(self):
for plugin in self.skel.plugins():
self.skel.create_symlinks(plugin)
class Skeleton(CMD):
"""
Work with the skeleton directories (create service templates)
"""
link = Link
class Run(CMD):
"""
Run a single operation
"""
arg_operation = Arg("operation", help="Python path to operation")
async def run(self):
# Push current directory into front of path so we can run things
# relative to where we are in the shell
sys.path.insert(0, os.getcwd())
# Lookup
modname, qualname_separator, qualname = self.operation.partition(":")
obj = importlib.import_module(modname)
if qualname_separator:
for attr in qualname.split("."):
obj = getattr(obj, attr)
if not cls.__doc__ is None:
print("%s:" % (cls.__qualname__))
print(cls.__doc__.rstrip())
else:
print("%s" % (cls.__qualname__))
print()
async def run(self):
"""
Display all classes registered with the entrypoint
"""
for cls in self.ENTRYPOINT.load():
self.display(cls)
class SourcesCMD(CMD):
arg_sources = Arg(
"-sources",
help="Sources for loading and saving",
nargs="+",
default=Sources(
JSONSource(
FileSourceConfig(
filename=os.path.join(
os.path.expanduser("~"), ".cache", "dffml.json"
)
)
)
),
type=BaseSource.load_labeled,
action=list_action(Sources),
MemoryInputSet,
MemoryInputSetConfig,
StringInputSetContext,
)
from ..config.config import BaseConfigLoader
from ..config.json import JSONConfigLoader
from ..source.source import SubsetSources
from ..util.data import merge
from ..util.entrypoint import load
from ..util.cli.arg import Arg
from ..util.cli.cmd import CMD
from ..util.cli.cmds import SourcesCMD, KeysCMD
from ..util.cli.parser import ParseInputsAction
class Merge(CMD):
arg_dataflows = Arg(
"dataflows", help="DataFlows to merge", nargs="+", type=pathlib.Path
)
arg_config = Arg(
"-config",
help="ConfigLoader to use for exporting",
type=BaseConfigLoader.load,
default=JSONConfigLoader,
)
arg_not_linked = Arg(
"-not-linked",
dest="not_linked",
help="Do not export dataflows as linked",
default=False,
action="store_true",
)
return CreateCMD
class Create(CMD):
"""
Create new models, operations, etc.
"""
model = create_from_skel("model")
operations = create_from_skel("operations")
service = create_from_skel("service")
source = create_from_skel("source")
config = create_from_skel("config")
class Link(CMD):
"""
Create required symlinks from skel/common to the other template directories
"""
skel = Skel()
async def run(self):
for plugin in self.skel.plugins():
self.skel.create_symlinks(plugin)
class Skeleton(CMD):
"""
Work with the skeleton directories (create service templates)
"""
class ModelCMD(CMD):
"""
Set a models model dir.
"""
arg_model = Arg(
"-model", help="Model used for ML", type=Model.load, required=True
)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.model = self.model.withconfig(self.extra_config)
class PortCMD(CMD):
arg_port = Arg("port", type=Port.load)
class KeysCMD(CMD):
arg_keys = Arg(
"-keys",
help="Key used for source lookup and evaluation",
nargs="+",
required=True,
)
class Service(CMD):
"""
Expose various functionalities of dffml
"""
pass
for i in pkg_resources.iter_entry_points("dffml.service.cli"):
loaded = i.load()
if issubclass(loaded, CMD):
setattr(Service, i.name, loaded)
return Service
class CLI(CMD):
"""
CLI interface for dffml
"""
version = Version
_list = List
edit = Edit
merge = Merge
_import = Import
export = Export
train = Train
accuracy = Accuracy
predict = Predict
service = services()
dataflow = Dataflow
config = Config