KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_schema method in tempest

Best Python code snippet using tempest_python

test_reflection.py

Source:test_reflection.py

1# coding: utf-82from sqlalchemy import exc3from sqlalchemy import FLOAT4from sqlalchemy import ForeignKey5from sqlalchemy import ForeignKeyConstraint6from sqlalchemy import Index7from sqlalchemy import inspect8from sqlalchemy import INTEGER9from sqlalchemy import Integer10from sqlalchemy import MetaData11from sqlalchemy import Numeric12from sqlalchemy import PrimaryKeyConstraint13from sqlalchemy import select14from sqlalchemy import String15from sqlalchemy import testing16from sqlalchemy import text17from sqlalchemy import Unicode18from sqlalchemy import UniqueConstraint19from sqlalchemy.dialects.oracle.base import BINARY_DOUBLE20from sqlalchemy.dialects.oracle.base import BINARY_FLOAT21from sqlalchemy.dialects.oracle.base import DOUBLE_PRECISION22from sqlalchemy.dialects.oracle.base import NUMBER23from sqlalchemy.testing import assert_raises24from sqlalchemy.testing import AssertsCompiledSQL25from sqlalchemy.testing import eq_26from sqlalchemy.testing import fixtures27from sqlalchemy.testing import is_28from sqlalchemy.testing.engines import testing_engine29from sqlalchemy.testing.schema import Column30from sqlalchemy.testing.schema import Table31class MultiSchemaTest(fixtures.TestBase, AssertsCompiledSQL):32    __only_on__ = "oracle"33    __backend__ = True34    @classmethod35    def setup_class(cls):36        # currently assuming full DBA privs for the user.37        # don't really know how else to go here unless38        # we connect as the other user.39        for stmt in (40            """41create table %(test_schema)s.parent(42    id integer primary key,43    data varchar2(50)44);45create table %(test_schema)s.child(46    id integer primary key,47    data varchar2(50),48    parent_id integer references %(test_schema)s.parent(id)49);50create table local_table(51    id integer primary key,52    data varchar2(50)53);54create synonym %(test_schema)s.ptable for %(test_schema)s.parent;55create synonym %(test_schema)s.ctable for %(test_schema)s.child;56create synonym %(test_schema)s_pt for %(test_schema)s.parent;57create synonym %(test_schema)s.local_table for local_table;58-- can't make a ref from local schema to the59-- remote schema's table without this,60-- *and* cant give yourself a grant !61-- so we give it to public.  ideas welcome.62grant references on %(test_schema)s.parent to public;63grant references on %(test_schema)s.child to public;64"""65            % {"test_schema": testing.config.test_schema}66        ).split(";"):67            if stmt.strip():68                testing.db.execute(stmt)69    @classmethod70    def teardown_class(cls):71        for stmt in (72            """73drop table %(test_schema)s.child;74drop table %(test_schema)s.parent;75drop table local_table;76drop synonym %(test_schema)s.ctable;77drop synonym %(test_schema)s.ptable;78drop synonym %(test_schema)s_pt;79drop synonym %(test_schema)s.local_table;80"""81            % {"test_schema": testing.config.test_schema}82        ).split(";"):83            if stmt.strip():84                testing.db.execute(stmt)85    @testing.provide_metadata86    def test_create_same_names_explicit_schema(self):87        schema = testing.db.dialect.default_schema_name88        meta = self.metadata89        parent = Table(90            "parent",91            meta,92            Column("pid", Integer, primary_key=True),93            schema=schema,94        )95        child = Table(96            "child",97            meta,98            Column("cid", Integer, primary_key=True),99            Column("pid", Integer, ForeignKey("%s.parent.pid" % schema)),100            schema=schema,101        )102        meta.create_all()103        parent.insert().execute({"pid": 1})104        child.insert().execute({"cid": 1, "pid": 1})105        eq_(child.select().execute().fetchall(), [(1, 1)])106    def test_reflect_alt_table_owner_local_synonym(self):107        meta = MetaData(testing.db)108        parent = Table(109            "%s_pt" % testing.config.test_schema,110            meta,111            autoload=True,112            oracle_resolve_synonyms=True,113        )114        self.assert_compile(115            parent.select(),116            "SELECT %(test_schema)s_pt.id, "117            "%(test_schema)s_pt.data FROM %(test_schema)s_pt"118            % {"test_schema": testing.config.test_schema},119        )120        select([parent]).execute().fetchall()121    def test_reflect_alt_synonym_owner_local_table(self):122        meta = MetaData(testing.db)123        parent = Table(124            "local_table",125            meta,126            autoload=True,127            oracle_resolve_synonyms=True,128            schema=testing.config.test_schema,129        )130        self.assert_compile(131            parent.select(),132            "SELECT %(test_schema)s.local_table.id, "133            "%(test_schema)s.local_table.data "134            "FROM %(test_schema)s.local_table"135            % {"test_schema": testing.config.test_schema},136        )137        select([parent]).execute().fetchall()138    @testing.provide_metadata139    def test_create_same_names_implicit_schema(self):140        meta = self.metadata141        parent = Table(142            "parent", meta, Column("pid", Integer, primary_key=True)143        )144        child = Table(145            "child",146            meta,147            Column("cid", Integer, primary_key=True),148            Column("pid", Integer, ForeignKey("parent.pid")),149        )150        meta.create_all()151        parent.insert().execute({"pid": 1})152        child.insert().execute({"cid": 1, "pid": 1})153        eq_(child.select().execute().fetchall(), [(1, 1)])154    def test_reflect_alt_owner_explicit(self):155        meta = MetaData(testing.db)156        parent = Table(157            "parent", meta, autoload=True, schema=testing.config.test_schema158        )159        child = Table(160            "child", meta, autoload=True, schema=testing.config.test_schema161        )162        self.assert_compile(163            parent.join(child),164            "%(test_schema)s.parent JOIN %(test_schema)s.child ON "165            "%(test_schema)s.parent.id = %(test_schema)s.child.parent_id"166            % {"test_schema": testing.config.test_schema},167        )168        select([parent, child]).select_from(169            parent.join(child)170        ).execute().fetchall()171    def test_reflect_local_to_remote(self):172        testing.db.execute(173            "CREATE TABLE localtable (id INTEGER "174            "PRIMARY KEY, parent_id INTEGER REFERENCES "175            "%(test_schema)s.parent(id))"176            % {"test_schema": testing.config.test_schema}177        )178        try:179            meta = MetaData(testing.db)180            lcl = Table("localtable", meta, autoload=True)181            parent = meta.tables["%s.parent" % testing.config.test_schema]182            self.assert_compile(183                parent.join(lcl),184                "%(test_schema)s.parent JOIN localtable ON "185                "%(test_schema)s.parent.id = "186                "localtable.parent_id"187                % {"test_schema": testing.config.test_schema},188            )189            select([parent, lcl]).select_from(190                parent.join(lcl)191            ).execute().fetchall()192        finally:193            testing.db.execute("DROP TABLE localtable")194    def test_reflect_alt_owner_implicit(self):195        meta = MetaData(testing.db)196        parent = Table(197            "parent", meta, autoload=True, schema=testing.config.test_schema198        )199        child = Table(200            "child", meta, autoload=True, schema=testing.config.test_schema201        )202        self.assert_compile(203            parent.join(child),204            "%(test_schema)s.parent JOIN %(test_schema)s.child "205            "ON %(test_schema)s.parent.id = "206            "%(test_schema)s.child.parent_id"207            % {"test_schema": testing.config.test_schema},208        )209        select([parent, child]).select_from(210            parent.join(child)211        ).execute().fetchall()212    def test_reflect_alt_owner_synonyms(self):213        testing.db.execute(214            "CREATE TABLE localtable (id INTEGER "215            "PRIMARY KEY, parent_id INTEGER REFERENCES "216            "%s.ptable(id))" % testing.config.test_schema217        )218        try:219            meta = MetaData(testing.db)220            lcl = Table(221                "localtable", meta, autoload=True, oracle_resolve_synonyms=True222            )223            parent = meta.tables["%s.ptable" % testing.config.test_schema]224            self.assert_compile(225                parent.join(lcl),226                "%(test_schema)s.ptable JOIN localtable ON "227                "%(test_schema)s.ptable.id = "228                "localtable.parent_id"229                % {"test_schema": testing.config.test_schema},230            )231            select([parent, lcl]).select_from(232                parent.join(lcl)233            ).execute().fetchall()234        finally:235            testing.db.execute("DROP TABLE localtable")236    def test_reflect_remote_synonyms(self):237        meta = MetaData(testing.db)238        parent = Table(239            "ptable",240            meta,241            autoload=True,242            schema=testing.config.test_schema,243            oracle_resolve_synonyms=True,244        )245        child = Table(246            "ctable",247            meta,248            autoload=True,249            schema=testing.config.test_schema,250            oracle_resolve_synonyms=True,251        )252        self.assert_compile(253            parent.join(child),254            "%(test_schema)s.ptable JOIN "255            "%(test_schema)s.ctable "256            "ON %(test_schema)s.ptable.id = "257            "%(test_schema)s.ctable.parent_id"258            % {"test_schema": testing.config.test_schema},259        )260        select([parent, child]).select_from(261            parent.join(child)262        ).execute().fetchall()263class ConstraintTest(fixtures.TablesTest):264    __only_on__ = "oracle"265    __backend__ = True266    run_deletes = None267    @classmethod268    def define_tables(cls, metadata):269        Table("foo", metadata, Column("id", Integer, primary_key=True))270    def test_oracle_has_no_on_update_cascade(self):271        bar = Table(272            "bar",273            self.metadata,274            Column("id", Integer, primary_key=True),275            Column(276                "foo_id", Integer, ForeignKey("foo.id", onupdate="CASCADE")277            ),278        )279        assert_raises(exc.SAWarning, bar.create)280        bat = Table(281            "bat",282            self.metadata,283            Column("id", Integer, primary_key=True),284            Column("foo_id", Integer),285            ForeignKeyConstraint(["foo_id"], ["foo.id"], onupdate="CASCADE"),286        )287        assert_raises(exc.SAWarning, bat.create)288    def test_reflect_check_include_all(self):289        insp = inspect(testing.db)290        eq_(insp.get_check_constraints("foo"), [])291        eq_(292            [293                rec["sqltext"]294                for rec in insp.get_check_constraints("foo", include_all=True)295            ],296            ['"ID" IS NOT NULL'],297        )298class SystemTableTablenamesTest(fixtures.TestBase):299    __only_on__ = "oracle"300    __backend__ = True301    def setup(self):302        testing.db.execute("create table my_table (id integer)")303        testing.db.execute(304            "create global temporary table my_temp_table (id integer)"305        )306        testing.db.execute(307            "create table foo_table (id integer) tablespace SYSTEM"308        )309    def teardown(self):310        testing.db.execute("drop table my_temp_table")311        testing.db.execute("drop table my_table")312        testing.db.execute("drop table foo_table")313    def test_table_names_no_system(self):314        insp = inspect(testing.db)315        eq_(insp.get_table_names(), ["my_table"])316    def test_temp_table_names_no_system(self):317        insp = inspect(testing.db)318        eq_(insp.get_temp_table_names(), ["my_temp_table"])319    def test_table_names_w_system(self):320        engine = testing_engine(options={"exclude_tablespaces": ["FOO"]})321        insp = inspect(engine)322        eq_(323            set(insp.get_table_names()).intersection(324                ["my_table", "foo_table"]325            ),326            set(["my_table", "foo_table"]),327        )328class DontReflectIOTTest(fixtures.TestBase):329    """test that index overflow tables aren't included in330    table_names."""331    __only_on__ = "oracle"332    __backend__ = True333    def setup(self):334        testing.db.execute(335            """336        CREATE TABLE admin_docindex(337                token char(20),338                doc_id NUMBER,339                token_frequency NUMBER,340                token_offsets VARCHAR2(2000),341                CONSTRAINT pk_admin_docindex PRIMARY KEY (token, doc_id))342            ORGANIZATION INDEX343            TABLESPACE users344            PCTTHRESHOLD 20345            OVERFLOW TABLESPACE users346        """347        )348    def teardown(self):349        testing.db.execute("drop table admin_docindex")350    def test_reflect_all(self):351        m = MetaData(testing.db)352        m.reflect()353        eq_(set(t.name for t in m.tables.values()), set(["admin_docindex"]))354class UnsupportedIndexReflectTest(fixtures.TestBase):355    __only_on__ = "oracle"356    __backend__ = True357    @testing.emits_warning("No column names")358    @testing.provide_metadata359    def test_reflect_functional_index(self):360        metadata = self.metadata361        Table(362            "test_index_reflect",363            metadata,364            Column("data", String(20), primary_key=True),365        )366        metadata.create_all()367        testing.db.execute(368            "CREATE INDEX DATA_IDX ON " "TEST_INDEX_REFLECT (UPPER(DATA))"369        )370        m2 = MetaData(testing.db)371        Table("test_index_reflect", m2, autoload=True)372def all_tables_compression_missing():373    try:374        testing.db.execute("SELECT compression FROM all_tables")375        if "Enterprise Edition" not in testing.db.scalar(376            "select * from v$version"377        ):378            return True379        return False380    except Exception:381        return True382def all_tables_compress_for_missing():383    try:384        testing.db.execute("SELECT compress_for FROM all_tables")385        if "Enterprise Edition" not in testing.db.scalar(386            "select * from v$version"387        ):388            return True389        return False390    except Exception:391        return True392class TableReflectionTest(fixtures.TestBase):393    __only_on__ = "oracle"394    __backend__ = True395    @testing.provide_metadata396    @testing.fails_if(all_tables_compression_missing)397    def test_reflect_basic_compression(self):398        metadata = self.metadata399        tbl = Table(400            "test_compress",401            metadata,402            Column("data", Integer, primary_key=True),403            oracle_compress=True,404        )405        metadata.create_all()406        m2 = MetaData(testing.db)407        tbl = Table("test_compress", m2, autoload=True)408        # Don't hardcode the exact value, but it must be non-empty409        assert tbl.dialect_options["oracle"]["compress"]410    @testing.provide_metadata411    @testing.fails_if(all_tables_compress_for_missing)412    def test_reflect_oltp_compression(self):413        metadata = self.metadata414        tbl = Table(415            "test_compress",416            metadata,417            Column("data", Integer, primary_key=True),418            oracle_compress="OLTP",419        )420        metadata.create_all()421        m2 = MetaData(testing.db)422        tbl = Table("test_compress", m2, autoload=True)423        assert tbl.dialect_options["oracle"]["compress"] == "OLTP"424class RoundTripIndexTest(fixtures.TestBase):425    __only_on__ = "oracle"426    __backend__ = True427    @testing.provide_metadata428    def test_basic(self):429        metadata = self.metadata430        s_table = Table(431            "sometable",432            metadata,433            Column("id_a", Unicode(255), primary_key=True),434            Column("id_b", Unicode(255), primary_key=True, unique=True),435            Column("group", Unicode(255), primary_key=True),436            Column("col", Unicode(255)),437            UniqueConstraint("col", "group"),438        )439        # "group" is a keyword, so lower case440        normalind = Index("tableind", s_table.c.id_b, s_table.c.group)441        Index(442            "compress1", s_table.c.id_a, s_table.c.id_b, oracle_compress=True443        )444        Index(445            "compress2",446            s_table.c.id_a,447            s_table.c.id_b,448            s_table.c.col,449            oracle_compress=1,450        )451        metadata.create_all()452        mirror = MetaData(testing.db)453        mirror.reflect()454        metadata.drop_all()455        mirror.create_all()456        inspect = MetaData(testing.db)457        inspect.reflect()458        def obj_definition(obj):459            return (460                obj.__class__,461                tuple([c.name for c in obj.columns]),462                getattr(obj, "unique", None),463            )464        # find what the primary k constraint name should be465        primaryconsname = testing.db.scalar(466            text(467                """SELECT constraint_name468               FROM all_constraints469               WHERE table_name = :table_name470               AND owner = :owner471               AND constraint_type = 'P' """472            ),473            table_name=s_table.name.upper(),474            owner=testing.db.dialect.default_schema_name.upper(),475        )476        reflectedtable = inspect.tables[s_table.name]477        # make a dictionary of the reflected objects:478        reflected = dict(479            [480                (obj_definition(i), i)481                for i in reflectedtable.indexes | reflectedtable.constraints482            ]483        )484        # assert we got primary key constraint and its name, Error485        # if not in dict486        assert (487            reflected[488                (PrimaryKeyConstraint, ("id_a", "id_b", "group"), None)489            ].name.upper()490            == primaryconsname.upper()491        )492        # Error if not in dict493        eq_(reflected[(Index, ("id_b", "group"), False)].name, normalind.name)494        assert (Index, ("id_b",), True) in reflected495        assert (Index, ("col", "group"), True) in reflected496        idx = reflected[(Index, ("id_a", "id_b"), False)]497        assert idx.dialect_options["oracle"]["compress"] == 2498        idx = reflected[(Index, ("id_a", "id_b", "col"), False)]499        assert idx.dialect_options["oracle"]["compress"] == 1500        eq_(len(reflectedtable.constraints), 1)501        eq_(len(reflectedtable.indexes), 5)502class DBLinkReflectionTest(fixtures.TestBase):503    __requires__ = ("oracle_test_dblink",)504    __only_on__ = "oracle"505    __backend__ = True506    @classmethod507    def setup_class(cls):508        from sqlalchemy.testing import config509        cls.dblink = config.file_config.get("sqla_testing", "oracle_db_link")510        # note that the synonym here is still not totally functional511        # when accessing via a different username as we do with the512        # multiprocess test suite, so testing here is minimal513        with testing.db.connect() as conn:514            conn.execute(515                "create table test_table "516                "(id integer primary key, data varchar2(50))"517            )518            conn.execute(519                "create synonym test_table_syn "520                "for test_table@%s" % cls.dblink521            )522    @classmethod523    def teardown_class(cls):524        with testing.db.connect() as conn:525            conn.execute("drop synonym test_table_syn")526            conn.execute("drop table test_table")527    def test_reflection(self):528        """test the resolution of the synonym/dblink. """529        m = MetaData()530        t = Table(531            "test_table_syn",532            m,533            autoload=True,534            autoload_with=testing.db,535            oracle_resolve_synonyms=True,536        )537        eq_(list(t.c.keys()), ["id", "data"])538        eq_(list(t.primary_key), [t.c.id])539class TypeReflectionTest(fixtures.TestBase):540    __only_on__ = "oracle"541    __backend__ = True542    @testing.provide_metadata543    def _run_test(self, specs, attributes):544        columns = [Column("c%i" % (i + 1), t[0]) for i, t in enumerate(specs)]545        m = self.metadata546        Table("oracle_types", m, *columns)547        m.create_all()548        m2 = MetaData(testing.db)549        table = Table("oracle_types", m2, autoload=True)550        for i, (reflected_col, spec) in enumerate(zip(table.c, specs)):551            expected_spec = spec[1]552            reflected_type = reflected_col.type553            is_(type(reflected_type), type(expected_spec))554            for attr in attributes:555                eq_(556                    getattr(reflected_type, attr),557                    getattr(expected_spec, attr),558                    "Column %s: Attribute %s value of %s does not "559                    "match %s for type %s"560                    % (561                        "c%i" % (i + 1),562                        attr,563                        getattr(reflected_type, attr),564                        getattr(expected_spec, attr),565                        spec[0],566                    ),567                )568    def test_integer_types(self):569        specs = [(Integer, INTEGER()), (Numeric, INTEGER())]570        self._run_test(specs, [])571    def test_number_types(self):572        specs = [(Numeric(5, 2), NUMBER(5, 2)), (NUMBER, NUMBER())]573        self._run_test(specs, ["precision", "scale"])574    def test_float_types(self):575        specs = [576            (DOUBLE_PRECISION(), FLOAT()),577            # when binary_precision is supported578            # (DOUBLE_PRECISION(), oracle.FLOAT(binary_precision=126)),579            (BINARY_DOUBLE(), BINARY_DOUBLE()),580            (BINARY_FLOAT(), BINARY_FLOAT()),581            (FLOAT(5), FLOAT()),582            # when binary_precision is supported583            # (FLOAT(5), oracle.FLOAT(binary_precision=5),),584            (FLOAT(), FLOAT()),585            # when binary_precision is supported586            # (FLOAT(5), oracle.FLOAT(binary_precision=126),),587        ]...

test_errors.py

Source:test_errors.py

1from __future__ import absolute_import2from __future__ import print_function3from .util import get_data4import unittest5from schema_salad.schema import load_schema, load_and_validate6from schema_salad.validate import ValidationException7from avro.schema import Names8import six9class TestErrors(unittest.TestCase):10    def test_errors(self):11        document_loader, avsc_names, schema_metadata, metaschema_loader = load_schema(12            get_data(u"tests/test_schema/CommonWorkflowLanguage.yml"))13        for t in ("test_schema/test1.cwl",14                  "test_schema/test2.cwl",15                  "test_schema/test3.cwl",16                  "test_schema/test4.cwl",17                  "test_schema/test5.cwl",18                  "test_schema/test6.cwl",19                  "test_schema/test7.cwl",20                  "test_schema/test8.cwl",21                  "test_schema/test9.cwl",22                  "test_schema/test10.cwl",23                  "test_schema/test11.cwl",24                  "test_schema/test12.cwl",25                  "test_schema/test13.cwl",26                  "test_schema/test14.cwl",27                  "test_schema/test15.cwl"):28            with self.assertRaises(ValidationException):29                try:30                    load_and_validate(document_loader, avsc_names,31                            six.text_type(get_data("tests/"+t)), True)32                except ValidationException as e:33                    print("\n", e)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.