Repository: quantopian/zipline
File: tests/pipeline/test_pipeline_algo.py (Python)

Code before:
        def handle_data(context, data):
            pass

        def before_trading_start(context, data):
            context.results = pipeline_output('test')
            self.assertTrue(context.results.empty)
            count[0] += 1

        self.run_algorithm(
            initialize=initialize,
            handle_data=handle_data,
            before_trading_start=before_trading_start,
            sim_params=SimulationParameters(
                start_session=self.dates[0],
                end_session=current_day,
                data_frequency='daily',
                emission_rate='daily',
                trading_calendar=self.trading_calendar,
            )
        )

        self.assertTrue(count[0] > 0)


class PipelineSequenceTestCase(
    WithSeededRandomPipelineEngine,
    WithMakeAlgo,
    ZiplineTestCase,
):

    @classmethod
    def init_class_fixtures(cls):
        super(PipelineSequenceTestCase, cls).init_class_fixtures()

    def make_algo_kwargs(self, **overrides):
        return self.merge_with_inherited_algo_kwargs(
            PipelineSequenceTestCase,
            suite_overrides=dict(
                get_pipeline_loader=lambda column: self.seeded_random_loader,
            ),
            method_overrides=overrides,
        )

    def test_pipeline_compute_before_bts(self):

        # for storing and keeping track of calls to BTS and TestFactor.compute
        history = []
        counter = [0]

        class TestFactor(CustomFactor, SingleInputMixin):


Review comment (suggestion):
Thinking about this with fresh eyes, I don't think you actually even need `WithSeededRandomPipelineEngine` here. You can just write a factor that doesn't take any inputs. (See notes below.)

Code after:
        def before_trading_start(context, data):
            context.results = pipeline_output('test')
            self.assertTrue(context.results.empty)
            count[0] += 1

        self.run_algorithm(
            initialize=initialize,
            handle_data=handle_data,
            before_trading_start=before_trading_start,
            sim_params=SimulationParameters(
                start_session=self.dates[0],
                end_session=current_day,
                data_frequency='daily',
                emission_rate='daily',
                trading_calendar=self.trading_calendar,
            )
        )

        self.assertTrue(count[0] > 0)


class PipelineSequenceTestCase(WithMakeAlgo, ZiplineTestCase):

    def make_algo_kwargs(self, **overrides):
        # run the algorithm for 3 days
        self.START_DATE = pd.Timestamp('2015-07-13', tz='utc')
        self.END_DATE = pd.Timestamp('2015-07-15', tz='utc')
        return self.merge_with_inherited_algo_kwargs(
            PipelineSequenceTestCase,
            suite_overrides=dict(
                get_pipeline_loader=lambda column: self.fake_pipeline,
            ),
            method_overrides=overrides,
        )

    def fake_pipeline(self):
        raise AssertionError

    def test_pipeline_compute_before_bts(self):

        # for storing and keeping track of calls to BTS and TestFactor.compute
        trace = []

        class TestFactor(CustomFactor):
            inputs = ()

            # window_length doesn't actually matter for this test case
            window_length = 1

            def compute(self, today, assets, out):
                trace.append("CustomFactor call")