code: def test_ingest_assets_versions(self):
      versions = (1, 2)
  
      called = [False]
  
      @self.register("bundle", create_writers=False)
      def bundle_ingest_no_create_writers(*args, **kwargs):
          called[0] = True
  
      now = pd.Timestamp.utcnow()
      with self.assertRaisesRegex(
          ValueError, "ingest .* creates writers .* downgrade"
      ):
          self.ingest(
              "bundle",
              self.environ,
              assets_versions=versions,
              timestamp=now - pd.Timedelta(seconds=1),
          )
      assert_false(called[0])
      assert_equal(
          len(ingestions_for_bundle("bundle", self.environ)), 1
      )
  
      @self.register("bundle", create_writers=True)
      def bundle_ingest_create_writers(
          environ,
          asset_db_writer,
          minute_bar_writer,
          daily_bar_writer,
          adjustment_writer,
          calendar,
          start_session,
          end_session,
          cache,
          show_progress,
          output_dir,
      ):
          self.assertIsNotNone(asset_db_writer)
          self.assertIsNotNone(minute_bar_writer)
          self.assertIsNotNone(daily_bar_writer)
          self.assertIsNotNone(adjustment_writer)
  
          equities = make_simple_equity_info(
              tuple(range(3)),
              self.START_DATE,
              self.END_DATE,
          )
          asset_db_writer.write(equities=equities)
          called[0] = True
  
      # Explicitly use different timestamp; otherwise, test could run so fast
      # that first ingestion is re-used.
      self.ingest(
          "bundle",
          self.environ,
          assets_versions=versions,
          timestamp=now,
      )
      assert_true(called[0])
  
      ingestions = ingestions_for_bundle("bundle", self.environ)
      assert_equal(len(ingestions), 2)
      for version in sorted(set(versions) | {ASSET_DB_VERSION}):
          eng = sa.create_engine(
              "sqlite:///"
              + asset_db_path(
                  "bundle",
                  to_bundle_ingest_dirname(
                      ingestions[0]
                  ),  # most recent
                  self.environ,
                  version,
              )
          )
          metadata = sa.MetaData()
          metadata.reflect(eng)
          version_table = metadata.tables["version_info"]
          check_version_info(eng, version_table, version)
  
imports: import os
  from nose_parameterized import parameterized
  import pandas
  import sqlalchemy
  from toolz import valmap
  import toolz.curried.operator
  from trading_calendars import TradingCalendar, get_calendar
  from zipline.assets import ASSET_DB_VERSION
  from zipline.assets.asset_writer import check_version_info
  from zipline.assets.synthetic import make_simple_equity_info
  from zipline.data.bundles import UnknownBundle, from_bundle_ingest_dirname, ingestions_for_bundle
  from zipline.data.bundles.core import _make_bundle_core, BadClean, to_bundle_ingest_dirname, asset_db_path
  from zipline.lib.adjustment import Float64Multiply
  from zipline.pipeline.loaders.synthetic import make_bar_data, expected_bar_values_2d
  from zipline.testing import subtest, str_to_seconds
  from zipline.testing.fixtures import WithInstanceTmpDir, ZiplineTestCase, WithDefaultDateBounds
  from zipline.testing.predicates import assert_equal, assert_false, assert_in, assert_is, assert_is_instance, assert_is_none, assert_raises, assert_true
  from zipline.utils.cache import dataframe_cache
  from zipline.utils.functional import apply
  import zipline.utils.paths
label_desc: Use image node_id 7 for calling the BundleCoreTestCase obj's underlying member method code with example usage: obj.test_ingest_assets_versions() without return types