test_migrations.py 文件源码

def test_upgrade_bulk(self):
        self.connection.execute(
            self.ip_policy.insert(),
            dict(id="1", size=None),
            dict(id="2", size=None))
        self.connection.execute(
            self.ip_policy_cidrs.insert(),
            dict(id="2", ip_policy_id="1", cidr="192.168.10.13/32"),
            dict(id="3", ip_policy_id="1", cidr="192.168.10.16/31"),
            dict(id="4", ip_policy_id="2", cidr="fd00::/64"))
        alembic_command.upgrade(self.config, '28e55acaf366')
        results = self.connection.execute(select([
            self.ip_policy])).fetchall()
        self.assertEqual(len(results), 2)
        for result in results:
            self.assertIn(result["id"], ("1", "2"))
            if result["id"] == "1":
                self.assertEqual(result["size"], 3)
            elif result["id"] == "2":
                self.assertEqual(result["size"], 2 ** 64)