1"""Test against the builders in the op.* module.""" 2 3from sqlalchemy import Boolean 4from sqlalchemy import CheckConstraint 5from sqlalchemy import Column 6from sqlalchemy import exc 7from sqlalchemy import ForeignKey 8from sqlalchemy import Index 9from sqlalchemy import Integer 10from sqlalchemy import MetaData 11from sqlalchemy import String 12from sqlalchemy import Table 13from sqlalchemy import UniqueConstraint 14from sqlalchemy.sql import column 15from sqlalchemy.sql import func 16from sqlalchemy.sql import text 17from sqlalchemy.sql.schema import quoted_name 18 19from alembic import op 20from alembic.operations import ops 21from alembic.operations import schemaobj 22from alembic.testing import assert_raises_message 23from alembic.testing import combinations 24from alembic.testing import config 25from alembic.testing import eq_ 26from alembic.testing import is_not_ 27from alembic.testing import mock 28from alembic.testing.fixtures import op_fixture 29from alembic.testing.fixtures import TestBase 30from alembic.util import sqla_compat 31 32 33class OpTest(TestBase): 34 def test_rename_table(self): 35 context = op_fixture() 36 op.rename_table("t1", "t2") 37 context.assert_("ALTER TABLE t1 RENAME TO t2") 38 39 def test_rename_table_schema(self): 40 context = op_fixture() 41 op.rename_table("t1", "t2", schema="foo") 42 context.assert_("ALTER TABLE foo.t1 RENAME TO foo.t2") 43 44 def test_create_index_arbitrary_expr(self): 45 context = op_fixture() 46 op.create_index("name", "tname", [func.foo(column("x"))]) 47 context.assert_("CREATE INDEX name ON tname (foo(x))") 48 49 def test_add_column_schema_hard_quoting(self): 50 51 context = op_fixture("postgresql") 52 op.add_column( 53 "somename", 54 Column("colname", String), 55 schema=quoted_name("some.schema", quote=True), 56 ) 57 58 context.assert_( 59 'ALTER TABLE "some.schema".somename ADD COLUMN colname VARCHAR' 60 ) 61 62 def test_rename_table_schema_hard_quoting(self): 63 64 context = op_fixture("postgresql") 65 op.rename_table( 66 "t1", "t2", schema=quoted_name("some.schema", quote=True) 67 ) 68 69 context.assert_('ALTER TABLE "some.schema".t1 RENAME TO t2') 70 71 def test_add_constraint_schema_hard_quoting(self): 72 73 context = op_fixture("postgresql") 74 op.create_check_constraint( 75 "ck_user_name_len", 76 "user_table", 77 func.len(column("name")) > 5, 78 schema=quoted_name("some.schema", quote=True), 79 ) 80 context.assert_( 81 'ALTER TABLE "some.schema".user_table ADD ' 82 "CONSTRAINT ck_user_name_len CHECK (len(name) > 5)" 83 ) 84 85 def test_create_index_quoting(self): 86 context = op_fixture("postgresql") 87 op.create_index("geocoded", "locations", ["IShouldBeQuoted"]) 88 context.assert_( 89 'CREATE INDEX geocoded ON locations ("IShouldBeQuoted")' 90 ) 91 92 def test_create_index_expressions(self): 93 context = op_fixture() 94 op.create_index("geocoded", "locations", [text("lower(coordinates)")]) 95 context.assert_( 96 "CREATE INDEX geocoded ON locations (lower(coordinates))" 97 ) 98 99 def test_add_column(self): 100 context = op_fixture() 101 op.add_column("t1", Column("c1", Integer, nullable=False)) 102 context.assert_("ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL") 103 104 def test_add_column_already_attached(self): 105 context = op_fixture() 106 c1 = Column("c1", Integer, nullable=False) 107 Table("t", MetaData(), c1) 108 109 op.add_column("t1", c1) 110 context.assert_("ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL") 111 112 def test_add_column_w_check(self): 113 context = op_fixture() 114 op.add_column( 115 "t1", 116 Column("c1", Integer, CheckConstraint("c1 > 5"), nullable=False), 117 ) 118 context.assert_( 119 "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL CHECK (c1 > 5)" 120 ) 121 122 def test_add_column_schema(self): 123 context = op_fixture() 124 op.add_column( 125 "t1", Column("c1", Integer, nullable=False), schema="foo" 126 ) 127 context.assert_("ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL") 128 129 def test_add_column_with_default(self): 130 context = op_fixture() 131 op.add_column( 132 "t1", Column("c1", Integer, nullable=False, server_default="12") 133 ) 134 context.assert_( 135 "ALTER TABLE t1 ADD COLUMN c1 INTEGER DEFAULT '12' NOT NULL" 136 ) 137 138 def test_add_column_with_index(self): 139 context = op_fixture() 140 op.add_column("t1", Column("c1", Integer, nullable=False, index=True)) 141 context.assert_( 142 "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL", 143 "CREATE INDEX ix_t1_c1 ON t1 (c1)", 144 ) 145 146 def test_add_column_schema_with_default(self): 147 context = op_fixture() 148 op.add_column( 149 "t1", 150 Column("c1", Integer, nullable=False, server_default="12"), 151 schema="foo", 152 ) 153 context.assert_( 154 "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER DEFAULT '12' NOT NULL" 155 ) 156 157 def test_add_column_fk(self): 158 context = op_fixture() 159 op.add_column( 160 "t1", Column("c1", Integer, ForeignKey("c2.id"), nullable=False) 161 ) 162 context.assert_( 163 "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL", 164 "ALTER TABLE t1 ADD FOREIGN KEY(c1) REFERENCES c2 (id)", 165 ) 166 167 def test_add_column_schema_fk(self): 168 context = op_fixture() 169 op.add_column( 170 "t1", 171 Column("c1", Integer, ForeignKey("c2.id"), nullable=False), 172 schema="foo", 173 ) 174 context.assert_( 175 "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL", 176 "ALTER TABLE foo.t1 ADD FOREIGN KEY(c1) REFERENCES c2 (id)", 177 ) 178 179 def test_add_column_schema_type(self): 180 """Test that a schema type generates its constraints....""" 181 context = op_fixture() 182 op.add_column( 183 "t1", Column("c1", Boolean(create_constraint=True), nullable=False) 184 ) 185 context.assert_( 186 "ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL", 187 "ALTER TABLE t1 ADD CHECK (c1 IN (0, 1))", 188 ) 189 190 def test_add_column_schema_schema_type(self): 191 """Test that a schema type generates its constraints....""" 192 context = op_fixture() 193 op.add_column( 194 "t1", 195 Column("c1", Boolean(create_constraint=True), nullable=False), 196 schema="foo", 197 ) 198 context.assert_( 199 "ALTER TABLE foo.t1 ADD COLUMN c1 BOOLEAN NOT NULL", 200 "ALTER TABLE foo.t1 ADD CHECK (c1 IN (0, 1))", 201 ) 202 203 def test_add_column_schema_type_checks_rule(self): 204 """Test that a schema type doesn't generate a 205 constraint based on check rule.""" 206 context = op_fixture("postgresql") 207 op.add_column( 208 "t1", Column("c1", Boolean(create_constraint=True), nullable=False) 209 ) 210 context.assert_("ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL") 211 212 def test_add_column_fk_self_referential(self): 213 context = op_fixture() 214 op.add_column( 215 "t1", Column("c1", Integer, ForeignKey("t1.c2"), nullable=False) 216 ) 217 context.assert_( 218 "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL", 219 "ALTER TABLE t1 ADD FOREIGN KEY(c1) REFERENCES t1 (c2)", 220 ) 221 222 def test_add_column_schema_fk_self_referential(self): 223 context = op_fixture() 224 op.add_column( 225 "t1", 226 Column("c1", Integer, ForeignKey("foo.t1.c2"), nullable=False), 227 schema="foo", 228 ) 229 context.assert_( 230 "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL", 231 "ALTER TABLE foo.t1 ADD FOREIGN KEY(c1) REFERENCES foo.t1 (c2)", 232 ) 233 234 def test_add_column_fk_schema(self): 235 context = op_fixture() 236 op.add_column( 237 "t1", 238 Column("c1", Integer, ForeignKey("remote.t2.c2"), nullable=False), 239 ) 240 context.assert_( 241 "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL", 242 "ALTER TABLE t1 ADD FOREIGN KEY(c1) REFERENCES remote.t2 (c2)", 243 ) 244 245 def test_add_column_schema_fk_schema(self): 246 context = op_fixture() 247 op.add_column( 248 "t1", 249 Column("c1", Integer, ForeignKey("remote.t2.c2"), nullable=False), 250 schema="foo", 251 ) 252 context.assert_( 253 "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL", 254 "ALTER TABLE foo.t1 ADD FOREIGN KEY(c1) REFERENCES remote.t2 (c2)", 255 ) 256 257 def test_drop_column(self): 258 context = op_fixture() 259 op.drop_column("t1", "c1") 260 context.assert_("ALTER TABLE t1 DROP COLUMN c1") 261 262 def test_drop_column_schema(self): 263 context = op_fixture() 264 op.drop_column("t1", "c1", schema="foo") 265 context.assert_("ALTER TABLE foo.t1 DROP COLUMN c1") 266 267 def test_alter_column_nullable(self): 268 context = op_fixture() 269 op.alter_column("t", "c", nullable=True) 270 context.assert_( 271 # TODO: not sure if this is PG only or standard 272 # SQL 273 "ALTER TABLE t ALTER COLUMN c DROP NOT NULL" 274 ) 275 276 def test_alter_column_schema_nullable(self): 277 context = op_fixture() 278 op.alter_column("t", "c", nullable=True, schema="foo") 279 context.assert_( 280 # TODO: not sure if this is PG only or standard 281 # SQL 282 "ALTER TABLE foo.t ALTER COLUMN c DROP NOT NULL" 283 ) 284 285 def test_alter_column_not_nullable(self): 286 context = op_fixture() 287 op.alter_column("t", "c", nullable=False) 288 context.assert_( 289 # TODO: not sure if this is PG only or standard 290 # SQL 291 "ALTER TABLE t ALTER COLUMN c SET NOT NULL" 292 ) 293 294 def test_alter_column_schema_not_nullable(self): 295 context = op_fixture() 296 op.alter_column("t", "c", nullable=False, schema="foo") 297 context.assert_( 298 # TODO: not sure if this is PG only or standard 299 # SQL 300 "ALTER TABLE foo.t ALTER COLUMN c SET NOT NULL" 301 ) 302 303 def test_alter_column_rename(self): 304 context = op_fixture() 305 op.alter_column("t", "c", new_column_name="x") 306 context.assert_("ALTER TABLE t RENAME c TO x") 307 308 def test_alter_column_schema_rename(self): 309 context = op_fixture() 310 op.alter_column("t", "c", new_column_name="x", schema="foo") 311 context.assert_("ALTER TABLE foo.t RENAME c TO x") 312 313 def test_alter_column_type(self): 314 context = op_fixture() 315 op.alter_column("t", "c", type_=String(50)) 316 context.assert_("ALTER TABLE t ALTER COLUMN c TYPE VARCHAR(50)") 317 318 def test_alter_column_schema_type(self): 319 context = op_fixture() 320 op.alter_column("t", "c", type_=String(50), schema="foo") 321 context.assert_("ALTER TABLE foo.t ALTER COLUMN c TYPE VARCHAR(50)") 322 323 def test_alter_column_set_default(self): 324 context = op_fixture() 325 op.alter_column("t", "c", server_default="q") 326 context.assert_("ALTER TABLE t ALTER COLUMN c SET DEFAULT 'q'") 327 328 def test_alter_column_schema_set_default(self): 329 context = op_fixture() 330 op.alter_column("t", "c", server_default="q", schema="foo") 331 context.assert_("ALTER TABLE foo.t ALTER COLUMN c SET DEFAULT 'q'") 332 333 def test_alter_column_set_compiled_default(self): 334 context = op_fixture() 335 op.alter_column( 336 "t", "c", server_default=func.utc_thing(func.current_timestamp()) 337 ) 338 context.assert_( 339 "ALTER TABLE t ALTER COLUMN c " 340 "SET DEFAULT utc_thing(CURRENT_TIMESTAMP)" 341 ) 342 343 def test_alter_column_schema_set_compiled_default(self): 344 context = op_fixture() 345 op.alter_column( 346 "t", 347 "c", 348 server_default=func.utc_thing(func.current_timestamp()), 349 schema="foo", 350 ) 351 context.assert_( 352 "ALTER TABLE foo.t ALTER COLUMN c " 353 "SET DEFAULT utc_thing(CURRENT_TIMESTAMP)" 354 ) 355 356 def test_alter_column_drop_default(self): 357 context = op_fixture() 358 op.alter_column("t", "c", server_default=None) 359 context.assert_("ALTER TABLE t ALTER COLUMN c DROP DEFAULT") 360 361 def test_alter_column_schema_drop_default(self): 362 context = op_fixture() 363 op.alter_column("t", "c", server_default=None, schema="foo") 364 context.assert_("ALTER TABLE foo.t ALTER COLUMN c DROP DEFAULT") 365 366 @combinations( 367 (lambda: sqla_compat.Computed("foo * 5"), lambda: None), 368 (lambda: None, lambda: sqla_compat.Computed("foo * 5")), 369 ( 370 lambda: sqla_compat.Computed("foo * 42"), 371 lambda: sqla_compat.Computed("foo * 5"), 372 ), 373 ) 374 @config.requirements.computed_columns_api 375 def test_alter_column_computed_not_supported(self, sd, esd): 376 op_fixture() 377 assert_raises_message( 378 exc.CompileError, 379 'Adding or removing a "computed" construct, e.g. ' 380 "GENERATED ALWAYS AS, to or from an existing column is not " 381 "supported.", 382 op.alter_column, 383 "t1", 384 "c1", 385 server_default=sd(), 386 existing_server_default=esd(), 387 ) 388 389 @combinations( 390 (lambda: sqla_compat.Identity(), lambda: None), 391 (lambda: None, lambda: sqla_compat.Identity()), 392 ( 393 lambda: sqla_compat.Identity(), 394 lambda: sqla_compat.Identity(), 395 ), 396 ) 397 @config.requirements.identity_columns_api 398 def test_alter_column_identity_not_supported(self, sd, esd): 399 op_fixture() 400 assert_raises_message( 401 exc.CompileError, 402 'Adding, removing or modifying an "identity" construct, ' 403 "e.g. GENERATED AS IDENTITY, to or from an existing " 404 "column is not supported in this dialect.", 405 op.alter_column, 406 "t1", 407 "c1", 408 server_default=sd(), 409 existing_server_default=esd(), 410 ) 411 412 def test_alter_column_schema_type_unnamed(self): 413 context = op_fixture("mssql", native_boolean=False) 414 op.alter_column("t", "c", type_=Boolean(create_constraint=True)) 415 context.assert_( 416 "ALTER TABLE t ALTER COLUMN c BIT", 417 "ALTER TABLE t ADD CHECK (c IN (0, 1))", 418 ) 419 420 def test_alter_column_schema_schema_type_unnamed(self): 421 context = op_fixture("mssql", native_boolean=False) 422 op.alter_column( 423 "t", "c", type_=Boolean(create_constraint=True), schema="foo" 424 ) 425 context.assert_( 426 "ALTER TABLE foo.t ALTER COLUMN c BIT", 427 "ALTER TABLE foo.t ADD CHECK (c IN (0, 1))", 428 ) 429 430 def test_alter_column_schema_type_named(self): 431 context = op_fixture("mssql", native_boolean=False) 432 op.alter_column( 433 "t", "c", type_=Boolean(name="xyz", create_constraint=True) 434 ) 435 context.assert_( 436 "ALTER TABLE t ALTER COLUMN c BIT", 437 "ALTER TABLE t ADD CONSTRAINT xyz CHECK (c IN (0, 1))", 438 ) 439 440 def test_alter_column_schema_schema_type_named(self): 441 context = op_fixture("mssql", native_boolean=False) 442 op.alter_column( 443 "t", 444 "c", 445 type_=Boolean(name="xyz", create_constraint=True), 446 schema="foo", 447 ) 448 context.assert_( 449 "ALTER TABLE foo.t ALTER COLUMN c BIT", 450 "ALTER TABLE foo.t ADD CONSTRAINT xyz CHECK (c IN (0, 1))", 451 ) 452 453 @combinations((True,), (False,), argnames="pass_existing_type") 454 @combinations((True,), (False,), argnames="change_nullability") 455 def test_generic_alter_column_type_and_nullability( 456 self, pass_existing_type, change_nullability 457 ): 458 # this test is also on the mssql dialect in test_mssql 459 context = op_fixture() 460 461 args = dict(type_=Integer) 462 if pass_existing_type: 463 args["existing_type"] = String(15) 464 465 if change_nullability: 466 args["nullable"] = False 467 468 op.alter_column("t", "c", **args) 469 470 if change_nullability: 471 context.assert_( 472 "ALTER TABLE t ALTER COLUMN c SET NOT NULL", 473 "ALTER TABLE t ALTER COLUMN c TYPE INTEGER", 474 ) 475 else: 476 context.assert_("ALTER TABLE t ALTER COLUMN c TYPE INTEGER") 477 478 def test_alter_column_schema_type_existing_type(self): 479 context = op_fixture("mssql", native_boolean=False) 480 op.alter_column( 481 "t", 482 "c", 483 type_=String(10), 484 existing_type=Boolean(name="xyz", create_constraint=True), 485 ) 486 context.assert_( 487 "ALTER TABLE t DROP CONSTRAINT xyz", 488 "ALTER TABLE t ALTER COLUMN c VARCHAR(10)", 489 ) 490 491 def test_alter_column_schema_schema_type_existing_type(self): 492 context = op_fixture("mssql", native_boolean=False) 493 op.alter_column( 494 "t", 495 "c", 496 type_=String(10), 497 existing_type=Boolean(name="xyz", create_constraint=True), 498 schema="foo", 499 ) 500 context.assert_( 501 "ALTER TABLE foo.t DROP CONSTRAINT xyz", 502 "ALTER TABLE foo.t ALTER COLUMN c VARCHAR(10)", 503 ) 504 505 def test_alter_column_schema_type_existing_type_no_const(self): 506 context = op_fixture("postgresql") 507 op.alter_column("t", "c", type_=String(10), existing_type=Boolean()) 508 context.assert_("ALTER TABLE t ALTER COLUMN c TYPE VARCHAR(10)") 509 510 def test_alter_column_schema_schema_type_existing_type_no_const(self): 511 context = op_fixture("postgresql") 512 op.alter_column( 513 "t", "c", type_=String(10), existing_type=Boolean(), schema="foo" 514 ) 515 context.assert_("ALTER TABLE foo.t ALTER COLUMN c TYPE VARCHAR(10)") 516 517 def test_alter_column_schema_type_existing_type_no_new_type(self): 518 context = op_fixture("postgresql") 519 op.alter_column("t", "c", nullable=False, existing_type=Boolean()) 520 context.assert_("ALTER TABLE t ALTER COLUMN c SET NOT NULL") 521 522 def test_alter_column_schema_schema_type_existing_type_no_new_type(self): 523 context = op_fixture("postgresql") 524 op.alter_column( 525 "t", "c", nullable=False, existing_type=Boolean(), schema="foo" 526 ) 527 context.assert_("ALTER TABLE foo.t ALTER COLUMN c SET NOT NULL") 528 529 def test_add_foreign_key(self): 530 context = op_fixture() 531 op.create_foreign_key( 532 "fk_test", "t1", "t2", ["foo", "bar"], ["bat", "hoho"] 533 ) 534 context.assert_( 535 "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " 536 "REFERENCES t2 (bat, hoho)" 537 ) 538 539 def test_add_foreign_key_schema(self): 540 context = op_fixture() 541 op.create_foreign_key( 542 "fk_test", 543 "t1", 544 "t2", 545 ["foo", "bar"], 546 ["bat", "hoho"], 547 source_schema="foo2", 548 referent_schema="bar2", 549 ) 550 context.assert_( 551 "ALTER TABLE foo2.t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " 552 "REFERENCES bar2.t2 (bat, hoho)" 553 ) 554 555 def test_add_foreign_key_schema_same_tablename(self): 556 context = op_fixture() 557 op.create_foreign_key( 558 "fk_test", 559 "t1", 560 "t1", 561 ["foo", "bar"], 562 ["bat", "hoho"], 563 source_schema="foo2", 564 referent_schema="bar2", 565 ) 566 context.assert_( 567 "ALTER TABLE foo2.t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " 568 "REFERENCES bar2.t1 (bat, hoho)" 569 ) 570 571 def test_add_foreign_key_onupdate(self): 572 context = op_fixture() 573 op.create_foreign_key( 574 "fk_test", 575 "t1", 576 "t2", 577 ["foo", "bar"], 578 ["bat", "hoho"], 579 onupdate="CASCADE", 580 ) 581 context.assert_( 582 "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " 583 "REFERENCES t2 (bat, hoho) ON UPDATE CASCADE" 584 ) 585 586 def test_add_foreign_key_ondelete(self): 587 context = op_fixture() 588 op.create_foreign_key( 589 "fk_test", 590 "t1", 591 "t2", 592 ["foo", "bar"], 593 ["bat", "hoho"], 594 ondelete="CASCADE", 595 ) 596 context.assert_( 597 "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " 598 "REFERENCES t2 (bat, hoho) ON DELETE CASCADE" 599 ) 600 601 def test_add_foreign_key_deferrable(self): 602 context = op_fixture() 603 op.create_foreign_key( 604 "fk_test", 605 "t1", 606 "t2", 607 ["foo", "bar"], 608 ["bat", "hoho"], 609 deferrable=True, 610 ) 611 context.assert_( 612 "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " 613 "REFERENCES t2 (bat, hoho) DEFERRABLE" 614 ) 615 616 def test_add_foreign_key_initially(self): 617 context = op_fixture() 618 op.create_foreign_key( 619 "fk_test", 620 "t1", 621 "t2", 622 ["foo", "bar"], 623 ["bat", "hoho"], 624 initially="deferred", 625 ) 626 context.assert_( 627 "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " 628 "REFERENCES t2 (bat, hoho) INITIALLY deferred" 629 ) 630 631 @config.requirements.foreign_key_match 632 def test_add_foreign_key_match(self): 633 context = op_fixture() 634 op.create_foreign_key( 635 "fk_test", 636 "t1", 637 "t2", 638 ["foo", "bar"], 639 ["bat", "hoho"], 640 match="SIMPLE", 641 ) 642 context.assert_( 643 "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " 644 "REFERENCES t2 (bat, hoho) MATCH SIMPLE" 645 ) 646 647 def test_add_foreign_key_dialect_kw(self): 648 op_fixture() 649 with mock.patch("sqlalchemy.schema.ForeignKeyConstraint") as fkc: 650 op.create_foreign_key( 651 "fk_test", 652 "t1", 653 "t2", 654 ["foo", "bar"], 655 ["bat", "hoho"], 656 foobar_arg="xyz", 657 ) 658 if config.requirements.foreign_key_match.enabled: 659 eq_( 660 fkc.mock_calls[0], 661 mock.call( 662 ["foo", "bar"], 663 ["t2.bat", "t2.hoho"], 664 onupdate=None, 665 ondelete=None, 666 name="fk_test", 667 foobar_arg="xyz", 668 deferrable=None, 669 initially=None, 670 match=None, 671 ), 672 ) 673 else: 674 eq_( 675 fkc.mock_calls[0], 676 mock.call( 677 ["foo", "bar"], 678 ["t2.bat", "t2.hoho"], 679 onupdate=None, 680 ondelete=None, 681 name="fk_test", 682 foobar_arg="xyz", 683 deferrable=None, 684 initially=None, 685 ), 686 ) 687 688 def test_add_foreign_key_self_referential(self): 689 context = op_fixture() 690 op.create_foreign_key("fk_test", "t1", "t1", ["foo"], ["bar"]) 691 context.assert_( 692 "ALTER TABLE t1 ADD CONSTRAINT fk_test " 693 "FOREIGN KEY(foo) REFERENCES t1 (bar)" 694 ) 695 696 def test_add_primary_key_constraint(self): 697 context = op_fixture() 698 op.create_primary_key("pk_test", "t1", ["foo", "bar"]) 699 context.assert_( 700 "ALTER TABLE t1 ADD CONSTRAINT pk_test PRIMARY KEY (foo, bar)" 701 ) 702 703 def test_add_primary_key_constraint_schema(self): 704 context = op_fixture() 705 op.create_primary_key("pk_test", "t1", ["foo"], schema="bar") 706 context.assert_( 707 "ALTER TABLE bar.t1 ADD CONSTRAINT pk_test PRIMARY KEY (foo)" 708 ) 709 710 def test_add_check_constraint(self): 711 context = op_fixture() 712 op.create_check_constraint( 713 "ck_user_name_len", "user_table", func.len(column("name")) > 5 714 ) 715 context.assert_( 716 "ALTER TABLE user_table ADD CONSTRAINT ck_user_name_len " 717 "CHECK (len(name) > 5)" 718 ) 719 720 def test_add_check_constraint_schema(self): 721 context = op_fixture() 722 op.create_check_constraint( 723 "ck_user_name_len", 724 "user_table", 725 func.len(column("name")) > 5, 726 schema="foo", 727 ) 728 context.assert_( 729 "ALTER TABLE foo.user_table ADD CONSTRAINT ck_user_name_len " 730 "CHECK (len(name) > 5)" 731 ) 732 733 def test_add_unique_constraint(self): 734 context = op_fixture() 735 op.create_unique_constraint("uk_test", "t1", ["foo", "bar"]) 736 context.assert_( 737 "ALTER TABLE t1 ADD CONSTRAINT uk_test UNIQUE (foo, bar)" 738 ) 739 740 def test_add_unique_constraint_schema(self): 741 context = op_fixture() 742 op.create_unique_constraint( 743 "uk_test", "t1", ["foo", "bar"], schema="foo" 744 ) 745 context.assert_( 746 "ALTER TABLE foo.t1 ADD CONSTRAINT uk_test UNIQUE (foo, bar)" 747 ) 748 749 def test_drop_constraint(self): 750 context = op_fixture() 751 op.drop_constraint("foo_bar_bat", "t1") 752 context.assert_("ALTER TABLE t1 DROP CONSTRAINT foo_bar_bat") 753 754 def test_drop_constraint_schema(self): 755 context = op_fixture() 756 op.drop_constraint("foo_bar_bat", "t1", schema="foo") 757 context.assert_("ALTER TABLE foo.t1 DROP CONSTRAINT foo_bar_bat") 758 759 def test_create_index(self): 760 context = op_fixture() 761 op.create_index("ik_test", "t1", ["foo", "bar"]) 762 context.assert_("CREATE INDEX ik_test ON t1 (foo, bar)") 763 764 def test_create_unique_index(self): 765 context = op_fixture() 766 op.create_index("ik_test", "t1", ["foo", "bar"], unique=True) 767 context.assert_("CREATE UNIQUE INDEX ik_test ON t1 (foo, bar)") 768 769 def test_create_index_quote_flag(self): 770 context = op_fixture() 771 op.create_index("ik_test", "t1", ["foo", "bar"], quote=True) 772 context.assert_('CREATE INDEX "ik_test" ON t1 (foo, bar)') 773 774 def test_create_index_table_col_event(self): 775 context = op_fixture() 776 777 op.create_index( 778 "ik_test", "tbl_with_auto_appended_column", ["foo", "bar"] 779 ) 780 context.assert_( 781 "CREATE INDEX ik_test ON tbl_with_auto_appended_column (foo, bar)" 782 ) 783 784 def test_add_unique_constraint_col_event(self): 785 context = op_fixture() 786 op.create_unique_constraint( 787 "ik_test", "tbl_with_auto_appended_column", ["foo", "bar"] 788 ) 789 context.assert_( 790 "ALTER TABLE tbl_with_auto_appended_column " 791 "ADD CONSTRAINT ik_test UNIQUE (foo, bar)" 792 ) 793 794 def test_create_index_schema(self): 795 context = op_fixture() 796 op.create_index("ik_test", "t1", ["foo", "bar"], schema="foo") 797 context.assert_("CREATE INDEX ik_test ON foo.t1 (foo, bar)") 798 799 def test_drop_index(self): 800 context = op_fixture() 801 op.drop_index("ik_test") 802 context.assert_("DROP INDEX ik_test") 803 804 def test_drop_index_schema(self): 805 context = op_fixture() 806 op.drop_index("ik_test", schema="foo") 807 context.assert_("DROP INDEX foo.ik_test") 808 809 def test_drop_table(self): 810 context = op_fixture() 811 op.drop_table("tb_test") 812 context.assert_("DROP TABLE tb_test") 813 814 def test_drop_table_schema(self): 815 context = op_fixture() 816 op.drop_table("tb_test", schema="foo") 817 context.assert_("DROP TABLE foo.tb_test") 818 819 def test_create_table_selfref(self): 820 context = op_fixture() 821 op.create_table( 822 "some_table", 823 Column("id", Integer, primary_key=True), 824 Column("st_id", Integer, ForeignKey("some_table.id")), 825 ) 826 context.assert_( 827 "CREATE TABLE some_table (" 828 "id INTEGER NOT NULL, " 829 "st_id INTEGER, " 830 "PRIMARY KEY (id), " 831 "FOREIGN KEY(st_id) REFERENCES some_table (id))" 832 ) 833 834 def test_create_table_check_constraint(self): 835 context = op_fixture() 836 t1 = op.create_table( 837 "some_table", 838 Column("id", Integer, primary_key=True), 839 Column("foo_id", Integer), 840 CheckConstraint("foo_id>5", name="ck_1"), 841 ) 842 context.assert_( 843 "CREATE TABLE some_table (" 844 "id INTEGER NOT NULL, " 845 "foo_id INTEGER, " 846 "PRIMARY KEY (id), " 847 "CONSTRAINT ck_1 CHECK (foo_id>5))" 848 ) 849 850 ck = [c for c in t1.constraints if isinstance(c, CheckConstraint)] 851 eq_(ck[0].name, "ck_1") 852 853 def test_create_table_unique_constraint(self): 854 context = op_fixture() 855 t1 = op.create_table( 856 "some_table", 857 Column("id", Integer, primary_key=True), 858 Column("foo_id", Integer), 859 UniqueConstraint("foo_id", name="uq_1"), 860 ) 861 context.assert_( 862 "CREATE TABLE some_table (" 863 "id INTEGER NOT NULL, " 864 "foo_id INTEGER, " 865 "PRIMARY KEY (id), " 866 "CONSTRAINT uq_1 UNIQUE (foo_id))" 867 ) 868 869 uq = [c for c in t1.constraints if isinstance(c, UniqueConstraint)] 870 eq_(uq[0].name, "uq_1") 871 872 def test_create_table_unique_flag(self): 873 context = op_fixture() 874 t1 = op.create_table( 875 "some_table", 876 Column("id", Integer, primary_key=True), 877 Column("foo_id", Integer, unique=True), 878 ) 879 context.assert_( 880 "CREATE TABLE some_table (id INTEGER NOT NULL, foo_id INTEGER, " 881 "PRIMARY KEY (id), UNIQUE (foo_id))" 882 ) 883 884 uq = [c for c in t1.constraints if isinstance(c, UniqueConstraint)] 885 assert uq 886 887 def test_create_table_index_flag(self): 888 context = op_fixture() 889 t1 = op.create_table( 890 "some_table", 891 Column("id", Integer, primary_key=True), 892 Column("foo_id", Integer, index=True), 893 ) 894 context.assert_( 895 "CREATE TABLE some_table (id INTEGER NOT NULL, foo_id INTEGER, " 896 "PRIMARY KEY (id))", 897 "CREATE INDEX ix_some_table_foo_id ON some_table (foo_id)", 898 ) 899 900 assert t1.indexes 901 902 def test_create_table_index(self): 903 context = op_fixture() 904 t1 = op.create_table( 905 "some_table", 906 Column("id", Integer, primary_key=True), 907 Column("foo_id", Integer), 908 Index("ix_1", "foo_id"), 909 ) 910 context.assert_( 911 "CREATE TABLE some_table (" 912 "id INTEGER NOT NULL, " 913 "foo_id INTEGER, " 914 "PRIMARY KEY (id))", 915 "CREATE INDEX ix_1 ON some_table (foo_id)", 916 ) 917 918 ix = list(t1.indexes) 919 eq_(ix[0].name, "ix_1") 920 921 def test_create_table_fk_and_schema(self): 922 context = op_fixture() 923 t1 = op.create_table( 924 "some_table", 925 Column("id", Integer, primary_key=True), 926 Column("foo_id", Integer, ForeignKey("foo.id")), 927 schema="schema", 928 ) 929 context.assert_( 930 "CREATE TABLE schema.some_table (" 931 "id INTEGER NOT NULL, " 932 "foo_id INTEGER, " 933 "PRIMARY KEY (id), " 934 "FOREIGN KEY(foo_id) REFERENCES foo (id))" 935 ) 936 eq_(t1.c.id.name, "id") 937 eq_(t1.schema, "schema") 938 939 def test_create_table_no_pk(self): 940 context = op_fixture() 941 t1 = op.create_table( 942 "some_table", 943 Column("x", Integer), 944 Column("y", Integer), 945 Column("z", Integer), 946 ) 947 context.assert_( 948 "CREATE TABLE some_table (x INTEGER, y INTEGER, z INTEGER)" 949 ) 950 assert not t1.primary_key 951 952 def test_create_table_two_fk(self): 953 context = op_fixture() 954 op.create_table( 955 "some_table", 956 Column("id", Integer, primary_key=True), 957 Column("foo_id", Integer, ForeignKey("foo.id")), 958 Column("foo_bar", Integer, ForeignKey("foo.bar")), 959 ) 960 context.assert_( 961 "CREATE TABLE some_table (" 962 "id INTEGER NOT NULL, " 963 "foo_id INTEGER, " 964 "foo_bar INTEGER, " 965 "PRIMARY KEY (id), " 966 "FOREIGN KEY(foo_id) REFERENCES foo (id), " 967 "FOREIGN KEY(foo_bar) REFERENCES foo (bar))" 968 ) 969 970 def test_inline_literal(self): 971 context = op_fixture() 972 from sqlalchemy.sql import table, column 973 from sqlalchemy import String, Integer 974 975 account = table( 976 "account", column("name", String), column("id", Integer) 977 ) 978 op.execute( 979 account.update() 980 .where(account.c.name == op.inline_literal("account 1")) 981 .values({"name": op.inline_literal("account 2")}) 982 ) 983 op.execute( 984 account.update() 985 .where(account.c.id == op.inline_literal(1)) 986 .values({"id": op.inline_literal(2)}) 987 ) 988 context.assert_( 989 "UPDATE account SET name='account 2' " 990 "WHERE account.name = 'account 1'", 991 "UPDATE account SET id=2 WHERE account.id = 1", 992 ) 993 994 def test_cant_op(self): 995 if hasattr(op, "_proxy"): 996 del op._proxy 997 assert_raises_message( 998 NameError, 999 "Can't invoke function 'inline_literal', as the " 1000 "proxy object has not yet been established " 1001 "for the Alembic 'Operations' class. " 1002 "Try placing this code inside a callable.", 1003 op.inline_literal, 1004 "asdf", 1005 ) 1006 1007 def test_naming_changes(self): 1008 context = op_fixture() 1009 op.alter_column("t", "c", new_column_name="x") 1010 context.assert_("ALTER TABLE t RENAME c TO x") 1011 1012 context = op_fixture("mysql") 1013 op.drop_constraint("f1", "t1", type_="foreignkey") 1014 context.assert_("ALTER TABLE t1 DROP FOREIGN KEY f1") 1015 1016 def test_naming_changes_drop_idx(self): 1017 context = op_fixture("mssql") 1018 op.drop_index("ik_test", table_name="t1") 1019 context.assert_("DROP INDEX ik_test ON t1") 1020 1021 @config.requirements.comments 1022 def test_create_table_comment_op(self): 1023 context = op_fixture() 1024 1025 op.create_table_comment("some_table", "table comment") 1026 1027 context.assert_("COMMENT ON TABLE some_table IS 'table comment'") 1028 1029 @config.requirements.comments 1030 def test_drop_table_comment_op(self): 1031 context = op_fixture() 1032 1033 op.drop_table_comment("some_table") 1034 1035 context.assert_("COMMENT ON TABLE some_table IS NULL") 1036 1037 1038class SQLModeOpTest(TestBase): 1039 def test_auto_literals(self): 1040 context = op_fixture(as_sql=True, literal_binds=True) 1041 from sqlalchemy.sql import table, column 1042 from sqlalchemy import String, Integer 1043 1044 account = table( 1045 "account", column("name", String), column("id", Integer) 1046 ) 1047 op.execute( 1048 account.update() 1049 .where(account.c.name == op.inline_literal("account 1")) 1050 .values({"name": op.inline_literal("account 2")}) 1051 ) 1052 op.execute(text("update table set foo=:bar").bindparams(bar="bat")) 1053 context.assert_( 1054 "UPDATE account SET name='account 2' " 1055 "WHERE account.name = 'account 1'", 1056 "update table set foo='bat'", 1057 ) 1058 1059 def test_create_table_literal_binds(self): 1060 context = op_fixture(as_sql=True, literal_binds=True) 1061 1062 op.create_table( 1063 "some_table", 1064 Column("id", Integer, primary_key=True), 1065 Column("st_id", Integer, ForeignKey("some_table.id")), 1066 ) 1067 1068 context.assert_( 1069 "CREATE TABLE some_table (id INTEGER NOT NULL, st_id INTEGER, " 1070 "PRIMARY KEY (id), FOREIGN KEY(st_id) REFERENCES some_table (id))" 1071 ) 1072 1073 1074class CustomOpTest(TestBase): 1075 def test_custom_op(self): 1076 from alembic.operations import Operations, MigrateOperation 1077 1078 @Operations.register_operation("create_sequence") 1079 class CreateSequenceOp(MigrateOperation): 1080 """Create a SEQUENCE.""" 1081 1082 def __init__(self, sequence_name, **kw): 1083 self.sequence_name = sequence_name 1084 self.kw = kw 1085 1086 @classmethod 1087 def create_sequence(cls, operations, sequence_name, **kw): 1088 """Issue a "CREATE SEQUENCE" instruction.""" 1089 1090 op = CreateSequenceOp(sequence_name, **kw) 1091 return operations.invoke(op) 1092 1093 @Operations.implementation_for(CreateSequenceOp) 1094 def create_sequence(operations, operation): 1095 operations.execute("CREATE SEQUENCE %s" % operation.sequence_name) 1096 1097 context = op_fixture() 1098 op.create_sequence("foob") 1099 context.assert_("CREATE SEQUENCE foob") 1100 1101 1102class ObjectFromToTest(TestBase): 1103 """Test operation round trips for to_obj() / from_obj(). 1104 1105 Previously, these needed to preserve the "original" item 1106 to this, but this makes them harder to work with. 1107 1108 As of #803 the constructs try to behave more intelligently 1109 about the state they were given, so that they can both "reverse" 1110 themselves but also take into accout their current state. 1111 1112 """ 1113 1114 def test_drop_index(self): 1115 schema_obj = schemaobj.SchemaObjects() 1116 idx = schema_obj.index("x", "y", ["z"]) 1117 op = ops.DropIndexOp.from_index(idx) 1118 is_not_(op.to_index(), idx) 1119 1120 def test_drop_index_add_kw(self): 1121 schema_obj = schemaobj.SchemaObjects() 1122 idx = schema_obj.index("x", "y", ["z"]) 1123 op = ops.DropIndexOp.from_index(idx) 1124 1125 op.kw["postgresql_concurrently"] = True 1126 eq_(op.to_index().dialect_kwargs["postgresql_concurrently"], True) 1127 1128 eq_( 1129 op.reverse().to_index().dialect_kwargs["postgresql_concurrently"], 1130 True, 1131 ) 1132 1133 def test_create_index(self): 1134 schema_obj = schemaobj.SchemaObjects() 1135 idx = schema_obj.index("x", "y", ["z"]) 1136 op = ops.CreateIndexOp.from_index(idx) 1137 1138 is_not_(op.to_index(), idx) 1139 1140 def test_create_index_add_kw(self): 1141 schema_obj = schemaobj.SchemaObjects() 1142 idx = schema_obj.index("x", "y", ["z"]) 1143 op = ops.CreateIndexOp.from_index(idx) 1144 1145 op.kw["postgresql_concurrently"] = True 1146 1147 eq_(op.to_index().dialect_kwargs["postgresql_concurrently"], True) 1148 eq_( 1149 op.reverse().to_index().dialect_kwargs["postgresql_concurrently"], 1150 True, 1151 ) 1152 1153 def test_drop_table(self): 1154 schema_obj = schemaobj.SchemaObjects() 1155 table = schema_obj.table( 1156 "x", 1157 Column("q", Integer), 1158 info={"custom": "value"}, 1159 prefixes=["FOREIGN"], 1160 postgresql_partition_by="x", 1161 comment="some comment", 1162 ) 1163 op = ops.DropTableOp.from_table(table) 1164 is_not_(op.to_table(), table) 1165 eq_(op.to_table().comment, table.comment) 1166 eq_(op.to_table().info, table.info) 1167 eq_(op.to_table()._prefixes, table._prefixes) 1168 1169 def test_drop_table_add_kw(self): 1170 schema_obj = schemaobj.SchemaObjects() 1171 table = schema_obj.table("x", Column("q", Integer)) 1172 op = ops.DropTableOp.from_table(table) 1173 1174 op.table_kw["postgresql_partition_by"] = "x" 1175 1176 eq_(op.to_table().dialect_kwargs["postgresql_partition_by"], "x") 1177 eq_( 1178 op.reverse().to_table().dialect_kwargs["postgresql_partition_by"], 1179 "x", 1180 ) 1181 1182 def test_create_table(self): 1183 schema_obj = schemaobj.SchemaObjects() 1184 table = schema_obj.table( 1185 "x", 1186 Column("q", Integer), 1187 postgresql_partition_by="x", 1188 prefixes=["FOREIGN"], 1189 info={"custom": "value"}, 1190 comment="some comment", 1191 ) 1192 op = ops.CreateTableOp.from_table(table) 1193 is_not_(op.to_table(), table) 1194 eq_(op.to_table().comment, table.comment) 1195 eq_(op.to_table().info, table.info) 1196 eq_(op.to_table()._prefixes, table._prefixes) 1197 1198 def test_create_table_add_kw(self): 1199 schema_obj = schemaobj.SchemaObjects() 1200 table = schema_obj.table("x", Column("q", Integer)) 1201 op = ops.CreateTableOp.from_table(table) 1202 op.kw["postgresql_partition_by"] = "x" 1203 1204 eq_(op.to_table().dialect_kwargs["postgresql_partition_by"], "x") 1205 eq_( 1206 op.reverse().to_table().dialect_kwargs["postgresql_partition_by"], 1207 "x", 1208 ) 1209 1210 def test_create_unique_constraint(self): 1211 schema_obj = schemaobj.SchemaObjects() 1212 const = schema_obj.unique_constraint("x", "foobar", ["a"]) 1213 op = ops.AddConstraintOp.from_constraint(const) 1214 is_not_(op.to_constraint(), const) 1215 1216 def test_create_unique_constraint_add_kw(self): 1217 schema_obj = schemaobj.SchemaObjects() 1218 const = schema_obj.unique_constraint("x", "foobar", ["a"]) 1219 op = ops.AddConstraintOp.from_constraint(const) 1220 is_not_(op.to_constraint(), const) 1221 1222 op.kw["sqlite_on_conflict"] = "IGNORE" 1223 1224 eq_(op.to_constraint().dialect_kwargs["sqlite_on_conflict"], "IGNORE") 1225 eq_( 1226 op.reverse().to_constraint().dialect_kwargs["sqlite_on_conflict"], 1227 "IGNORE", 1228 ) 1229 1230 def test_drop_unique_constraint(self): 1231 schema_obj = schemaobj.SchemaObjects() 1232 const = schema_obj.unique_constraint("x", "foobar", ["a"]) 1233 op = ops.DropConstraintOp.from_constraint(const) 1234 is_not_(op.to_constraint(), const) 1235 1236 def test_drop_unique_constraint_change_name(self): 1237 schema_obj = schemaobj.SchemaObjects() 1238 const = schema_obj.unique_constraint("x", "foobar", ["a"]) 1239 op = ops.DropConstraintOp.from_constraint(const) 1240 1241 op.constraint_name = "my_name" 1242 eq_(op.to_constraint().name, "my_name") 1243 eq_(op.reverse().to_constraint().name, "my_name") 1244 1245 def test_drop_constraint_not_available(self): 1246 op = ops.DropConstraintOp("x", "y", type_="unique") 1247 assert_raises_message( 1248 ValueError, "constraint cannot be produced", op.to_constraint 1249 ) 1250