How to use dffml - 10 common examples

def test_init(self):
        class CMDTest(CMD):
            arg_nope_present = Arg("nope", default=False)
            arg_ignored = Arg("ignored")

        cmd = CMDTest(nope=True)
        self.assertTrue(getattr(cmd, "log", False))
        self.assertTrue(getattr(cmd, "nope", False))

outputs={"add": is_add, "mult": is_mult, "numbers": numbers},
)
async def parse_line(line: str):
    return {
        "add": "add" in line,
        "mult": "mult" in line,
        "numbers": [int(item) for item in line.split() if item.isdigit()],
    }


OPIMPS = opimp_in(sys.modules[__name__])
OPERATIONS = operation_in(sys.modules[__name__])
DATAFLOW = DataFlow.auto(*OPIMPS)


class TestMemoryKeyValueStore(AsyncTestCase):
    def setUp(self):
        self.kvStore = MemoryKeyValueStore(BaseConfig())

    async def test_get_set(self):
        async with self.kvStore as kvstore:
            async with kvstore() as ctx:
                await ctx.set("feed", b"face")
                self.assertEqual(await ctx.get("feed"), b"face")

    async def test_get_none(self):
        async with self.kvStore as kvstore:
            async with kvstore() as ctx:
                self.assertEqual(await ctx.get("feed"), None)


class TestMemoryOperationImplementationNetwork(AsyncTestCase):

"operations",
            [
                ("{import_name}", "definitions.py"),
                ("{import_name}", "operations.py"),
                ("tests", "test_operations.py"),
            ],
        )

    async def test_service(self):
        await self.generic_test(
            "service",
            [("{import_name}", "misc.py"), ("tests", "test_service.py")],
        )


class TestDevelopSkelLink(AsyncTestCase):

    skel = Skel()

    async def test_run(self):
        # Skip if not in development mode
        if not is_develop("dffml"):
            self.skipTest("dffml not installed in development mode")
        await Develop.cli("skel", "link")
        common_files = [
            path.relative_to(self.skel.common)
            for path in self.skel.common_files()
        ]
        # At time of writing there are 4 plugins in skel/ change this as needed
        plugins = self.skel.plugins()
        self.assertGreater(len(plugins), 3)
        for plugin in plugins:

@op(
    inputs={"line": calc_string},
    outputs={"add": is_add, "mult": is_mult, "numbers": numbers},
)
async def parse_line(line: str):
    return {
        "add": "add" in line,
        "mult": "mult" in line,
        "numbers": [int(item) for item in line.split() if item.isdigit()],
    }


OPIMPS = opimp_in(sys.modules[__name__])
OPERATIONS = operation_in(sys.modules[__name__])
DATAFLOW = DataFlow.auto(*OPIMPS)


class TestMemoryKeyValueStore(AsyncTestCase):
    def setUp(self):
        self.kvStore = MemoryKeyValueStore(BaseConfig())

    async def test_get_set(self):
        async with self.kvStore as kvstore:
            async with kvstore() as ctx:
                await ctx.set("feed", b"face")
                self.assertEqual(await ctx.get("feed"), b"face")

    async def test_get_none(self):
        async with self.kvStore as kvstore:
            async with kvstore() as ctx:
                self.assertEqual(await ctx.get("feed"), None)

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)

    async def predict(self, repos: AsyncIterator[Repo]) -> AsyncIterator[Repo]:
        async for repo in repos:
            repo.predicted(random.random(), float(repo.src_url))
            yield repo

def setUp(self):
        self.null = Repo("null")
        self.full = Repo(
            "full",
            data=dict(
                features=dict(dead="beef"),
                extra=dict(extra="read all about it"),
            ),
            extra=dict(half=True),
        )

async def test_update(self):
        full_src_url = "0"
        empty_src_url = "1"
        full_repo = Repo(
            full_src_url,
            data={
                "features": {
                    "PetalLength": 3.9,
                    "PetalWidth": 1.2,
                    "SepalLength": 5.8,
                    "SepalWidth": 2.7,
                },
                "prediction": {"value": "feedface", "confidence": 0.42},
            },
        )
        empty_repo = Repo(
            empty_src_url,
            data={
                "features": {
                    "PetalLength": 3.9,