test_pipeline.py 文件源码

def testForce(self, existsMock, accessMock, subprocessMock):
        """
        If force=True is given to SlurmPipeline, SP_FORCE must be set to '1'
        in the step execution environment.
        """
        subprocessMock.return_value = ''

        sp = SlurmPipeline(
            {
                'steps': [
                    {
                        'name': 'name1',
                        'script': 'script1',
                    },
                ],
            })
        sp.schedule(force=True)

        subprocessMock.assert_has_calls([
            call(['script1'], cwd='.', universal_newlines=True,
                 stdin=DEVNULL, env=ANY),
        ])

        env = subprocessMock.mock_calls[0][2]['env']
        self.assertEqual('1', env['SP_FORCE'])