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

Code before:
                self.trading_days,
                self.ASSET_FINDER_COUNTRY_CODE,
            ),
        )
        result = self.run_pipeline(
            pipe, self.trading_days[-5], self.trading_days[-1]
        )

        # We should have one maximum every day.
        maxes_per_day = result.groupby(level=0)['maximum'].sum()
        self.assertTrue((maxes_per_day == 1).all())

        # The maximum computed by pipeline should match the maximum computed by
        # doing a groupby in pandas.
        groupby_max = result.groupby(level=0).factor.max()
        pipeline_max = (result.factor[result.maximum]
                        .reset_index(level=1, drop=True))

        assert_equal(groupby_max, pipeline_max)


class ResolveDomainTestCase(zf.ZiplineTestCase):

    def test_resolve_domain(self):
        engine_generic = SimplePipelineEngine(
            None, None, default_domain=GENERIC
        )
        engine_jp = SimplePipelineEngine(
            None, None, default_domain=JP_EQUITIES
        )

        pipe_generic = Pipeline()
        pipe_us = Pipeline(domain=US_EQUITIES)

        # the engine should resolve a pipeline that already has a domain
        # to that domain
        self.assertIs(
            engine_jp.resolve_domain(pipe_us),
            US_EQUITIES
        )

        # the engine should resolve a pipeline without a domain to the engine's
        # default
        self.assertIs(
            engine_jp.resolve_domain(pipe_generic),
            JP_EQUITIES
        )

        # a generic engine should resolve to the pipeline's domain
        # if it has one
        self.assertIs(

Review comment (suggestion):
I'd probably use `NamedExplodingObject` for these so that we raise a clear error if something in the engine starts depending on the asset finder or countries in loader dispatcher at construction time.

Code after:
            domain=EquitySessionDomain(
                self.trading_days,
                self.ASSET_FINDER_COUNTRY_CODE,
            ),
        )
        result = self.run_pipeline(
            pipe, self.trading_days[-5], self.trading_days[-1]
        )

        # We should have one maximum every day.
        maxes_per_day = result.groupby(level=0)['maximum'].sum()
        self.assertTrue((maxes_per_day == 1).all())

        # The maximum computed by pipeline should match the maximum computed by
        # doing a groupby in pandas.
        groupby_max = result.groupby(level=0).factor.max()
        pipeline_max = (result.factor[result.maximum]
                        .reset_index(level=1, drop=True))

        assert_equal(groupby_max, pipeline_max)


class ResolveDomainTestCase(zf.ZiplineTestCase):

    def test_resolve_domain(self):
        # we need to pass a get_loader and an asset_finder to construct
        # SimplePipelineEngine, but do not expect to use them
        get_loader = NamedExplodingObject(
            'self._get_loader',
            'SimplePipelineEngine does not currently depend on get_loader '
            'at construction time. Update this test if it now does.'
        )
        asset_finder = NamedExplodingObject(
            'self._finder',
            'SimplePipelineEngine does not currently depend on asset_finder '
            'at construction time. Update this test if it now does.'
        )

        engine_generic = SimplePipelineEngine(
            get_loader, asset_finder, default_domain=GENERIC
        )
        engine_jp = SimplePipelineEngine(
            get_loader, asset_finder, default_domain=JP_EQUITIES
        )

        pipe_generic = Pipeline()
        pipe_us = Pipeline(domain=US_EQUITIES)

        # the engine should resolve a pipeline that already has a domain
        # to that domain
        self.assertIs(