libs/dmigrations/tests/migration_state.py

   1 from dmigrations.tests.common import *
   2 from dmigrations.migration_state import MigrationState
   3 from dmigrations.migration_db import MigrationDb
   4
   5 create_new = """
   6 CREATE TABLE `dmigrations` (
   7   `id` int(11) NOT NULL auto_increment,
   8   `migration` VARCHAR(255) NOT NULL,
   9   PRIMARY KEY (`id`)
  10 ) ENGINE=InnoDB DEFAULT CHARSET=utf8
  11 """
  12
  13 class MigrationStateTest(TestCase):
  14   def set_up(self):
  15     from django.db import connection
  16     self.cursor = connection.cursor()
  17
  18     try: self.cursor.execute("DROP TABLE dmigrations")
  19     except: pass
  20
  21   def test_table_present_methods(self):
  22     si = MigrationState()
  23
  24     self.assert_equal(False, si.migration_table_present())
  25
  26     self.cursor.execute(create_new)
  27
  28     self.assert_equal(True, si.migration_table_present())
  29
  30   def test_init_on_fresh_db_creates_new_table(self):
  31     si = MigrationState()
  32
  33     self.assert_equal(False, si.migration_table_present())
  34
  35     si.init()
  36
  37     self.assert_equal(True, si.migration_table_present())
  38
  39   def test_init_on_initialized_db_does_nothing(self):
  40     self.cursor.execute(create_new)
  41
  42     si = MigrationState()
  43
  44     self.assert_equal(True, si.migration_table_present())
  45
  46     si.init()
  47
  48     self.assert_equal(True, si.migration_table_present())
  49
  50   def test_applying_and_unapplying(self):
  51     def assert_applied(f, b, h):
  52       self.assert_equal([f, b, h], [si.is_applied('001_foo'), si.is_applied('002_bar'), si.is_applied('005_hello')])
  53
  54     si = MigrationState()
  55     self.cursor.execute(create_new)
  56
  57     assert_applied(False, False, False)
  58
  59     si.mark_as_applied('001_foo')
  60     assert_applied(True, False, False)
  61
  62     si.mark_as_applied('005_hello')
  63     assert_applied(True, False, True)
  64
  65     si.mark_as_applied('005_hello')
  66     assert_applied(True, False, True)
  67
  68     si.mark_as_unapplied('002_bar')
  69     assert_applied(True, False, True)
  70
  71     si.mark_as_unapplied('005_hello')
  72     assert_applied(True, False, False)
  73
  74     si.mark_as_applied('002_bar')
  75     assert_applied(True, True, False)
  76
  77   def assert_plans(self, si, *plans):
  78     while plans:
  79       query, expected_plan, plans = plans[0], plans[1], plans[2:]
  80
  81       if isinstance(expected_plan, type(Exception)) and issubclass(expected_plan, Exception):
  82         self.assert_raises(expected_plan, lambda: si.plan(*query))
  83       else:
  84         actual_plan =  si.plan(*query)
  85         self.assert_equal(expected_plan, actual_plan, u"Plan for %s expected to be %s but was %s" % (query, expected_plan, actual_plan))
  86
  87   def test_plan(self):
  88     db = MigrationDb(migrations = ['001_foo', '002_bar', '005_omg', '005_hello'])
  89     si = MigrationState(migration_db=db)
  90
  91     si.init()
  92
  93     self.assert_plans(si,
  94       ['apply', '1'],         [('001_foo', 'up')],
  95       ['apply', '001_foo'],   [('001_foo', 'up')],
  96       ['apply', '1', '2'],    [('001_foo', 'up'), ('002_bar', 'up')],
  97       ['unapply', '001_foo'], [],
  98       ['unapply', '1', '2'],  [],
  99
 100       ['all'],                [('001_foo', 'up'), ('002_bar', 'up'), ('005_hello', 'up'), ('005_omg', 'up')],
 101       ['up'],                 [('001_foo', 'up')],
 102       ['down'],               [],
 103
 104       ['to', '1'],            [('001_foo', 'up')],
 105       ['to', '2'],            [('001_foo', 'up'), ('002_bar', 'up')],
 106       ['upto', '2'],          [('001_foo', 'up'), ('002_bar', 'up')],
 107       ['upto', '002_bar'],    [('001_foo', 'up'), ('002_bar', 'up')],
 108       ['upto', '4'],          [('001_foo', 'up'), ('002_bar', 'up')],
 109       ['upto', '004_foo'],    NoSuchMigrationError,
 110       ['upto', '5'],          AmbiguousMigrationNameError,
 111       ['upto', '005_hello'],  [('001_foo', 'up'), ('002_bar', 'up'), ('005_hello', 'up')],
 112       ['upto', '005_omg'],    [('001_foo', 'up'), ('002_bar', 'up'), ('005_hello', 'up'), ('005_omg', 'up')],
 113       ['downto', '1'],        [],
 114       ['downto', '2'],        [],
 115     )
 116
 117     si.mark_as_applied('002_bar')
 118
 119     self.assert_plans(si,
 120       ['apply', '1'],         [('001_foo', 'up')],
 121       ['apply', '001_foo'],   [('001_foo', 'up')],
 122       ['apply', '1', '2'],    [('001_foo', 'up')],
 123       ['unapply', '001_foo'], [],
 124       ['unapply', '1', '2'],  [('002_bar', 'down')],
 125
 126       ['all'],                [('001_foo', 'up'), ('005_hello', 'up'), ('005_omg', 'up')],
 127       ['up'],                 [('001_foo', 'up')],
 128       ['down'],               [('002_bar', 'down')],
 129
 130       ['to', '1'],            [('002_bar', 'down'), ('001_foo', 'up')],
 131       ['to', '2'],            [('001_foo', 'up')],
 132       ['upto', '2'],          [('001_foo', 'up')],
 133       ['upto', '002_bar'],    [('001_foo', 'up')],
 134       ['upto', '4'],          [('001_foo', 'up')],
 135       ['upto', '004_foo'],    NoSuchMigrationError,
 136       ['upto', '5'],          AmbiguousMigrationNameError,
 137       ['upto', '005_hello'],  [('001_foo', 'up'), ('005_hello', 'up')],
 138       ['upto', '005_omg'],    [('001_foo', 'up'), ('005_hello', 'up'), ('005_omg', 'up')],
 139       ['downto', '1'],        [('002_bar', 'down')],
 140       ['downto', '2'],        [],
 141     )
 142
 143     si.mark_as_applied('001_foo')
 144
 145     self.assert_plans(si,
 146       ['apply', '1'],         [],
 147       ['apply', '001_foo'],   [],
 148       ['apply', '1', '2'],    [],
 149       ['unapply', '001_foo'], [('001_foo', 'down')],
 150       ['unapply', '1', '2'],  [('001_foo', 'down'), ('002_bar', 'down')],
 151
 152       ['all'],                [('005_hello', 'up'), ('005_omg', 'up')],
 153       ['up'],                 [('005_hello', 'up')],
 154       ['down'],               [('002_bar', 'down')],
 155
 156       ['to', '1'],            [('002_bar', 'down')],
 157       ['to', '2'],            [],
 158       ['upto', '2'],          [],
 159       ['upto', '002_bar'],    [],
 160       ['upto', '4'],          [],
 161       ['upto', '004_foo'],    NoSuchMigrationError,
 162       ['upto', '5'],          AmbiguousMigrationNameError,
 163       ['upto', '005_hello'],  [('005_hello', 'up')],
 164       ['upto', '005_omg'],    [('005_hello', 'up'), ('005_omg', 'up')],
 165       ['downto', '1'],        [('002_bar', 'down')],
 166       ['downto', '2'],        [],
 167     )
 168
 169     si.mark_as_unapplied('002_bar')
 170
 171     self.assert_plans(si,
 172       ['apply', '1'],         [],
 173       ['apply', '001_foo'],   [],
 174       ['apply', '1', '2'],    [('002_bar', 'up')],
 175       ['unapply', '001_foo'], [('001_foo', 'down')],
 176       ['unapply', '1', '2'],  [('001_foo', 'down')],
 177
 178       ['all'],                [('002_bar', 'up'), ('005_hello', 'up'), ('005_omg', 'up')],
 179       ['up'],                 [('002_bar', 'up')],
 180       ['down'],               [('001_foo', 'down')],
 181
 182       ['to', '1'],            [],
 183       ['to', '2'],            [('002_bar', 'up')],
 184       ['upto', '2'],          [('002_bar', 'up')],
 185       ['upto', '002_bar'],    [('002_bar', 'up')],
 186       ['upto', '4'],          [('002_bar', 'up')],
 187       ['upto', '004_foo'],    NoSuchMigrationError,
 188       ['upto', '5'],          AmbiguousMigrationNameError,
 189       ['upto', '005_hello'],  [('002_bar', 'up'), ('005_hello', 'up')],
 190       ['upto', '005_omg'],    [('002_bar', 'up'), ('005_hello', 'up'), ('005_omg', 'up')],
 191       ['downto', '1'],        [],
 192       ['downto', '2'],        [],
 193     )
 194
 195     si.mark_as_applied('002_bar')
 196     si.mark_as_applied('005_hello')
 197     si.mark_as_applied('005_omg')
 198
 199     self.assert_plans(si,
 200       ['apply', '1'],         [],
 201       ['apply', '001_foo'],   [],
 202       ['apply', '1', '2'],    [],
 203       ['unapply', '001_foo'], [('001_foo', 'down')],
 204       ['unapply', '1', '2'],  [('001_foo', 'down'), ('002_bar', 'down')],
 205
 206       ['all'],                [],
 207       ['up'],                 [],
 208       ['down'],               [('005_omg', 'down')],
 209
 210       ['to', '1'],            [('005_omg', 'down'), ('005_hello', 'down'), ('002_bar', 'down')],
 211       ['to', '2'],            [('005_omg', 'down'), ('005_hello', 'down')],
 212       ['upto', '2'],          [],
 213       ['upto', '002_bar'],    [],
 214       ['upto', '4'],          [],
 215       ['upto', '004_foo'],    NoSuchMigrationError,
 216       ['upto', '5'],          AmbiguousMigrationNameError,
 217       ['upto', '005_hello'],  [],
 218       ['upto', '005_omg'],    [],
 219       ['downto', '1'],        [('005_omg', 'down'), ('005_hello', 'down'), ('002_bar', 'down')],
 220       ['downto', '2'],        [('005_omg', 'down'), ('005_hello', 'down')],
 221     )
 222
 223   def test_plan_with_dev(self):
 224     # NOTE: si.mark_as_applied modifies global state, not si (MigrationState) object.
 225     #       There is only one meaningful MigrationState object per database,
 226     #       so it's an almost-singleton unless we support multiple databases at the same time.
 227     #       This makes testing quite ugly.
 228     db = MigrationDb(migrations = ['001_foo', '002_DEV_bar', '005_omg', '006_DEV_hello'])
 229
 230     si = MigrationState(migration_db=db)
 231     si.init()
 232     self.assert_plans(si,
 233       ['up'],        [('001_foo', 'up')],
 234       ['upto', '5'], [('001_foo', 'up'), ('005_omg', 'up')],
 235       ['upto', '6'], [('001_foo', 'up'), ('005_omg', 'up')],
 236       ['all'],       [('001_foo', 'up'), ('005_omg', 'up')],
 237     )
 238
 239     si = MigrationState(migration_db=db, dev=True)
 240     si.init()
 241     self.assert_plans(si,
 242       ['up'],        [('001_foo', 'up')],
 243       ['upto', '5'], [('001_foo', 'up'), ('002_DEV_bar', 'up'), ('005_omg', 'up')],
 244       ['upto', '6'], [('001_foo', 'up'), ('002_DEV_bar', 'up'), ('005_omg', 'up'), ('006_DEV_hello', 'up')],
 245       ['all'],       [('001_foo', 'up'), ('002_DEV_bar', 'up'), ('005_omg', 'up'), ('006_DEV_hello', 'up')],
 246     )
 247
 248     si.mark_as_applied('001_foo')
 249
 250     si = MigrationState(migration_db=db)
 251     si.init()
 252     self.assert_plans(si,
 253       ['up'],        [('005_omg', 'up')],
 254       ['upto', '5'], [('005_omg', 'up')],
 255       ['upto', '6'], [('005_omg', 'up')],
 256       ['all'],       [('005_omg', 'up')],
 257     )
 258
 259     si = MigrationState(migration_db=db, dev=True)
 260     si.init()
 261     self.assert_plans(si,
 262       ['up'],        [('002_DEV_bar', 'up')],
 263       ['upto', '5'], [('002_DEV_bar', 'up'), ('005_omg', 'up')],
 264       ['upto', '6'], [('002_DEV_bar', 'up'), ('005_omg', 'up'), ('006_DEV_hello', 'up')],
 265       ['all'],       [('002_DEV_bar', 'up'), ('005_omg', 'up'), ('006_DEV_hello', 'up')],
 266     )
 267
 268     si.mark_as_applied('002_DEV_bar')
 269
 270     si = MigrationState(migration_db=db)
 271     si.init()
 272     self.assert_plans(si,
 273       ['up'],        [('005_omg', 'up')],
 274       ['upto', '5'], [('005_omg', 'up')],
 275       ['upto', '6'], [('005_omg', 'up')],
 276       ['all'],       [('005_omg', 'up')],
 277       ['down'],      [('001_foo', 'down')], # Is it better to always assume --dev on down ?
 278     )
 279
 280     si = MigrationState(migration_db=db, dev=True)
 281     si.init()
 282     self.assert_plans(si,
 283       ['up'],        [('005_omg', 'up')],
 284       ['upto', '5'], [('005_omg', 'up')],
 285       ['upto', '6'], [('005_omg', 'up'), ('006_DEV_hello', 'up')],
 286       ['all'],       [('005_omg', 'up'), ('006_DEV_hello', 'up')],
 287       ['down'],      [('002_DEV_bar', 'down')],
 288     )
 289
 290   def test_all_migrations_applied(self):
 291     db = MigrationDb(migrations = ['001_foo', '002_DEV_bar', '005_omg', '006_DEV_hello'])
 292     si = MigrationState(migration_db=db)
 293     si.init()
 294
 295     si.mark_as_applied('005_omg')
 296     si.mark_as_applied('001_foo')
 297     si.mark_as_applied('009_bogus')
 298
 299     self.assert_equal(['001_foo', '005_omg', '009_bogus'], si.all_migrations_applied())