test_migrations.py 文件源码

def test_upgrade_with_subnets_default_ip_policy_cidrs(self):
        self.connection.execute(
            self.subnets.insert(),
            dict(id="000", _cidr="192.168.10.0/24", ip_policy_id="111"))
        dt = datetime.datetime(1970, 1, 1)
        self.connection.execute(
            self.ip_policy_cidrs.insert(),
            dict(id="222", created_at=dt,
                 ip_policy_id="111", cidr="192.168.10.0/32"),
            dict(id="223", created_at=dt,
                 ip_policy_id="111", cidr="192.168.10.255/32"))
        alembic_command.upgrade(self.config, '45a07fac3d38')
        results = self.connection.execute(
            select([self.ip_policy_cidrs])).fetchall()
        self.assertEqual(len(results), 2)
        default_cidrs = ["192.168.10.0/32", "192.168.10.255/32"]
        self.assertIn(results[0]["cidr"], default_cidrs)
        self.assertIn(results[1]["cidr"], default_cidrs)
        self.assertTrue(results[0]["id"] == "222" or results[0]["id"] == "223")
        self.assertTrue(results[1]["id"] == "222" or results[1]["id"] == "223")
        self.assertEqual(results[0]["created_at"], dt)
        self.assertEqual(results[1]["created_at"], dt)