dffml: df: Add aenter and aexit to all

Signed-off-by: John Andersen <john.s.andersen@intel.com>

Author: John Andersen

.travis.yml

@@ -16,31 +16,35 @@ env:
   - PLUGIN=.
   - PLUGIN=model/tensorflow
   - PLUGIN=feature/git
+  - PLUGIN=feature/codesec
   - CHANGELOG=1
   - WHITESPACE=1
 cache:
   directories:
     - /home/travis/.cache/
     - /home/travis/.cargo/
+    - /home/travis/.local/
 before_install:
 - mkdir -p /home/travis/.cache/
 - pip install coverage codecov
 - |
-  curl -o /home/travis/.cache/cloc -sSL https://github.com/AlDanial/cloc/raw/1.80/cloc
-  sudo cp /home/travis/.cache/cloc /usr/bin/cloc
-  sudo chmod 755 /usr/bin/cloc
-- |
-  curl https://sh.rustup.rs -sSf | sh -s -- -y
-  source "$HOME/.cargo/env"
-  export tokeisrc=/home/travis/.cache/tokeisrc/
-  mkdir -p $tokeisrc
   if [ ! -f "$HOME/.cargo/bin/tokei" ]; then
+    curl https://sh.rustup.rs -sSf | sh -s -- -y
+    source "$HOME/.cargo/env"
+    export tokeisrc=/home/travis/.cache/tokeisrc/
+    mkdir -p $tokeisrc
     git clone https://github.com/Aaronepower/tokei.git --depth 1 $tokeisrc
     cd $tokeisrc
     cargo build --release
     cd -
     mv $tokeisrc/target/release/tokei "$HOME/.cargo/bin/tokei"
   fi
+- |
+  export clocpath="$HOME/.cargo/bin/cloc"
+  mkdir -p $(dirname $clocpath)
+  if [ ! -f "$clocpath" ]; then
+    curl -o $clocpath -sSL https://github.com/AlDanial/cloc/raw/1.80/cloc
+  fi
 - ''
 script:
 - ./.ci/run.sh

dffml/cli.py

@@ -159,75 +159,63 @@ class OperationsAll(OperationsCMD):
 
     # async def operations(self, sources, features):
     async def run_operations(self, sources):
-        # Orchestrate the running of these operations
-        async with self.dff(
-                input_network = self.input_network(
-                    self.input_network.config(self)),
-                operation_network = self.operation_network(
-                    self.operation_network.config(self)),
-                lock_network = self.lock_network(
-                    self.lock_network.config(self)),
-                rchecker = self.rchecker(
-                    self.rchecker.config(self)),
-                opimpn = self.opimpn(self.opimpn.config(self)),
-                orchestrator = self.orchestrator(
-                    self.orchestrator.config(self))
-                ) as dffctx:
-
-            # Create the inputs for the ouput operations
-            output_specs = [Input(value=value,
-                                 definition=self.definitions[def_name],
-                                 parents=False) \
-                for value, def_name in self.output_specs]
-
-            # Add our inputs to the input network with the context being the
-            # repo src_url
-            async for repo in sources.repos():
-                inputs = []
-                for value, def_name in self.inputs:
-                    inputs.append(Input(value=value,
-                                        definition=self.definitions[def_name],
-                                        parents=False))
-                if self.repo_def:
-                    inputs.append(Input(value=repo.src_url,
-                                        definition=self.definitions[self.repo_def],
-                                        parents=False))
-
-                await dffctx.ictx.add(
-                    MemoryInputSet(
-                        MemoryInputSetConfig(
-                            ctx=StringInputSetContext(repo.src_url),
-                            inputs=inputs + output_specs
+        async with self.dff as dff:
+            # Orchestrate the running of these operations
+            async with dff() as dffctx:
+                # Create the inputs for the ouput operations
+                output_specs = [Input(value=value,
+                                     definition=self.definitions[def_name],
+                                     parents=False) \
+                    for value, def_name in self.output_specs]
+
+                # Add our inputs to the input network with the context being the
+                # repo src_url
+                async for repo in sources.repos():
+                    inputs = []
+                    for value, def_name in self.inputs:
+                        inputs.append(Input(value=value,
+                                            definition=self.definitions[def_name],
+                                            parents=False))
+                    if self.repo_def:
+                        inputs.append(Input(value=repo.src_url,
+                                            definition=self.definitions[self.repo_def],
+                                            parents=False))
+
+                    await dffctx.ictx.add(
+                        MemoryInputSet(
+                            MemoryInputSetConfig(
+                                ctx=StringInputSetContext(repo.src_url),
+                                inputs=inputs + output_specs
+                            )
                         )
                     )
-                )
-
-            async for ctx, results in dffctx.evaluate():
-                ctx_str = (await ctx.handle()).as_string()
-                # TODO Make a RepoInputSetContext which would let us store the
-                # repo instead of recalling it by the URL
-                repo = await sources.repo(ctx_str)
-                # Remap the output operations to their feature
-                remap = {}
-                for output_operation_name, sub, feature_name in self.remap:
-                    if not output_operation_name in results:
-                        self.logger.error('[%s] results do not contain %s: %s',
-                                          ctx_str,
-                                          output_operation_name,
-                                          results)
-                        continue
-                    if not sub in results[output_operation_name]:
-                        self.logger.error('[%s] %s does not contain: %s',
-                                          ctx_str,
-                                          sub,
-                                          results[output_operation_name])
-                        continue
-                    remap[feature_name] = results[output_operation_name][sub]
-                # Store the results
-                repo.evaluated(remap)
-                yield repo
-                if self.update:
-                    await sources.update(repo)
+
+                async for ctx, results in dffctx.evaluate(strict=True):
+                    ctx_str = (await ctx.handle()).as_string()
+                    # TODO Make a RepoInputSetContext which would let us store the
+                    # repo instead of recalling it by the URL
+                    repo = await sources.repo(ctx_str)
+                    # Remap the output operations to their feature
+                    remap = {}
+                    for output_operation_name, sub, feature_name in self.remap:
+                        if not output_operation_name in results:
+                            self.logger.error('[%s] results do not contain %s: %s',
+                                              ctx_str,
+                                              output_operation_name,
+                                              results)
+                            continue
+                        if not sub in results[output_operation_name]:
+                            self.logger.error('[%s] %s does not contain: %s',
+                                              ctx_str,
+                                              sub,
+                                              results[output_operation_name])
+                            continue
+                        remap[feature_name] = results[output_operation_name][sub]
+                    # Store the results
+                    repo.evaluated(remap)
+                    yield repo
+                    if self.update:
+                        await sources.update(repo)
 
     async def run(self):
         # async with self.sources as sources, self.features as features:

dffml/df/base.py

@@ -24,6 +24,10 @@ def args(cls) -> Dict[str, Arg]:
     def config(cls, cmd: CMD):
         pass
 
+    @classmethod
+    def withconfig(cls, cmd: CMD):
+        return cls(cls.config(cmd))
+
 class OperationImplementationContext(abc.ABC):
 
     def __init__(self,
@@ -81,7 +85,10 @@ async def __call__(self,
                        ctx: 'BaseInputSetContext',
                        ictx: 'BaseInputNetworkContext') \
             -> OperationImplementationContext:
-        return OperationImplementationContext(self, ictx)
+        return OperationImplementationContext(self, ctx, ictx)
+
+    async def __aenter__(self) -> 'OperationImplementation':
+        return self
 
     async def __aexit__(self, exc_type, exc_value, traceback):
         pass
@@ -647,20 +654,20 @@ def __init__(self,
         self.logger = LOGGER.getChild(self.__class__.__qualname__)
 
     @abc.abstractmethod
-    async def run_operations(self) \
+    async def run_operations(self, strict: bool = False) \
             -> AsyncIterator[Tuple[BaseContextHandle, Dict[str, Any]]]:
         '''
         Run all Stage.PROCESSING operations
         '''
         pass
 
-    async def run_until_complete(self) \
+    async def run_until_complete(self, strict: bool = False) \
             -> AsyncIterator[Tuple[BaseContextHandle, Dict[str, Any]]]:
         '''
         Run all the operations then run cleanup and output operations
         '''
         # Run all operations until no more are run
-        async for ctx, results in self.run_operations():
+        async for ctx, results in self.run_operations(strict=strict):
             yield ctx, results
 
     async def __aenter__(self) -> 'BaseOrchestratorContext':

dffml/df/dff.py

@@ -16,22 +16,22 @@ def __init__(self,
                  operation_network: BaseOperationNetwork,
                  lock_network: BaseLockNetwork,
                  rchecker: BaseRedundancyChecker,
-                 opimpn: BaseOperationImplementationNetwork,
+                 opimp_network: BaseOperationImplementationNetwork,
                  orchestrator: BaseOrchestrator) -> None:
         self.input_network = input_network
         self.operation_network = operation_network
         self.lock_network = lock_network
         self.rchecker = rchecker
-        self.opimp_network = opimpn
+        self.opimp_network = opimp_network
         self.orchestrator = orchestrator
         self.logger = LOGGER.getChild(self.__class__.__qualname__)
         self.__stack = None
 
-    async def evaluate(self):
+    async def evaluate(self, strict=False):
         # Orchestrate the running of these operations
         async with self.orchestrator(self.ictx, self.octx, self.lctx, self.nctx,
                                      self.rctx) as orchestrate:
-            async for ctx, results in orchestrate.run_until_complete():
+            async for ctx, results in orchestrate.run_until_complete(strict=strict):
                 yield ctx, results
 
     async def __aenter__(self) -> 'DataFlowFacilitatorContext':
@@ -42,15 +42,15 @@ async def __aenter__(self) -> 'DataFlowFacilitatorContext':
         self.__stack = AsyncExitStack()
         await self.__stack.__aenter__()
         self.rctx = await self.__stack.enter_async_context(
-                self.rchecker())
+                self.rchecker)
         self.ictx = await self.__stack.enter_async_context(
-                self.input_network())
+                self.input_network)
         self.octx = await self.__stack.enter_async_context(
-                self.operation_network())
+                self.operation_network)
         self.lctx = await self.__stack.enter_async_context(
-                self.lock_network())
+                self.lock_network)
         self.nctx = await self.__stack.enter_async_context(
-                self.opimp_network())
+                self.opimp_network)
         return self
 
     async def __aexit__(self, exc_type, exc_value, traceback):
@@ -61,20 +61,45 @@ class DataFlowFacilitator(object):
     Data Flow Facilitator-tots
     '''
 
-    def __init__(self) -> None:
-        self.logger = LOGGER.getChild(self.__class__.__qualname__)
-
-    def __call__(self,
+    def __init__(self,
                  input_network: BaseInputNetwork,
                  operation_network: BaseOperationNetwork,
                  lock_network: BaseLockNetwork,
                  rchecker: BaseRedundancyChecker,
-                 opimpn: BaseOperationImplementationNetwork,
-                 orchestrator: BaseOrchestrator) \
-                         -> 'DataFlowFacilitatorContext':
-        return DataFlowFacilitatorContext(input_network,
-                                          operation_network,
-                                          lock_network,
-                                          rchecker,
-                                          opimpn,
-                                          orchestrator)
+                 opimp_network: BaseOperationImplementationNetwork,
+                 orchestrator: BaseOrchestrator) -> None:
+        self.rchecker = rchecker
+        self.input_network = input_network
+        self.operation_network = operation_network
+        self.lock_network = lock_network
+        self.opimp_network = opimp_network
+        self.orchestrator = orchestrator
+        self.logger = LOGGER.getChild(self.__class__.__qualname__)
+
+    def __call__(self) -> 'DataFlowFacilitatorContext':
+        return DataFlowFacilitatorContext(self.input_network(),
+                                          self.operation_network(),
+                                          self.lock_network(),
+                                          self.rchecker(),
+                                          self.opimp_network(),
+                                          self.orchestrator)
+
+    async def __aenter__(self) -> 'DataFlowFacilitator':
+        self.__stack = AsyncExitStack()
+        await self.__stack.__aenter__()
+        # self.rchecker = await self.__stack.enter_async_context(
+        #         self.rchecker)
+        # self.input_network = await self.__stack.enter_async_context(
+        #         self.input_network)
+        # self.operation_network = await self.__stack.enter_async_context(
+        #         self.operation_network)
+        # self.lock_network = await self.__stack.enter_async_context(
+        #         self.lock_network)
+        self.opimp_network = await self.__stack.enter_async_context(
+                self.opimp_network)
+        # self.orchestrator = await self.__stack.enter_async_context(
+        #         self.orchestrator)
+        return self
+
+    async def __aexit__(self, exc_type, exc_value, traceback):
+        await self.__stack.aclose()