U ^g%i*@sddlZddlZddlZddlZddlmZddlmZddlm Z ddlm Z ddlm Z ddlm Z dd l m Z dd l mZd d Zd2ddZddZddZddZd3ddZd4ddZd5ddZd6d!d"Zd#d$Zd%d&Zd7d(d)Zd*d+Zd,d-Zd.d/Zd0d1ZdS)8N)config) provision)util)command)script)Script)ScriptDirectorycCstjrdtjSdSdS)Nz scratch_%sZscratch)rZFOLLOWER_IDENTr r R/var/www/html/api-medvista/venv/lib/python3.8/site-packages/alembic/testing/env.py_get_staging_directorys r TgenericFcCst}|rtjtd}tj|r2td|tj|||d|rzt |dWnt k rjYnX|dks|t|t tj|dd|krdndt j|}|S)Nscriptsz2staging directory %s already exists; poor cleanup?)templateenv.py)Zpep3147_envonlysimplepep3147_everythingpep3147r)_testing_configospathjoinr existsAssertionErrorrinitrZload_python_fileAttributeErrormake_sourcelessrr from_config)creater sourcelesscfgrscr r r staging_envs*   r#cCs(ddlm}|jttddS)NrenginesT)sqlalchemy.testingr%Ztesting_reaperZ close_allshutilrmtreer r$r r r clear_staging_env:s  r)c CsBtjtd}tj|d}t|d}||W5QRXdS)Nrzscript.py.makow)rrrr openwrite)txtdir_rfr r r script_file_fixtureAs r0c Csbtjtd}d|}tj|d}t|}|r>t|t|d}||W5QRXdS)Nrz6 from alembic import context config = context.config rr*) rrrr rpyc_file_from_pathunlinkr+r,)r-r.rpyc_pathr/r r r env_file_fixtureHs    r4foo.dbcKs>tjtd}d||f}|r.tjr.||d<tj|||dS)Nrzsqlite:///%s/%sscope)urlfutureoptions)rrrr rZsqla_14 testing_utilZtesting_engine)tempnamer8r6r9r.r7r r r _sqlite_file_db\s   r<cCsNtjtd}d|}|p(dtjjjk}td|||r:dnd|rDdndfS) Nrsqlite:///%s/foo.dbr8a [alembic] script_location = %s sqlalchemy.url = %s sourceless = %s %s [loggers] keys = root,sqlalchemy [handlers] keys = console [logger_root] level = WARNING handlers = console qualname = [logger_sqlalchemy] level = DEBUG handlers = qualname = sqlalchemy.engine [handler_console] class = StreamHandler args = (sys.stderr,) level = NOTSET formatter = generic [formatters] keys = generic [formatter_generic] format = %%(levelname)-5.5s [%%(name)s] %%(message)s datefmt = %%H:%%M:%%S truefalsezsqlalchemy.future = true rrrr rdb __class__ __module___write_config_file)r r8r.r7sqlalchemy_futurer r r _sqlite_testing_configds%  rGr@cCsLtjtd}dtjjjk}d|}td|||r6dnd|r@dnd|fS)Nrr8r=a [alembic] script_location = %s sqlalchemy.url = %s sqlalchemy.future = %s sourceless = %s version_locations = %%(here)s/model1/ %%(here)s/model2/ %%(here)s/model3/ %s [loggers] keys = root [handlers] keys = console [logger_root] level = WARNING handlers = console qualname = [handler_console] class = StreamHandler args = (sys.stderr,) level = NOTSET formatter = generic [formatters] keys = generic [formatter_generic] format = %%(levelname)-5.5s [%%(name)s] %%(message)s datefmt = %%H:%%M:%%S r>r?rA)r Zextra_version_locationr.rFr7r r r _multi_dir_testing_configs!  rH postgresqlcCs"tjtd}td|||fS)zLuse a postgresql url with no host so that connections guaranteed to failra [alembic] script_location = %s sqlalchemy.url = %s:// %s [loggers] keys = root [handlers] keys = console [logger_root] level = WARNING handlers = console qualname = [handler_console] class = StreamHandler args = (sys.stderr,) level = NOTSET formatter = generic [formatters] keys = generic [formatter_generic] format = %%(levelname)-5.5s [%%(name)s] %%(message)s datefmt = %%H:%%M:%%S )rrrr rE)dialectZ directivesr.r r r _no_sql_testing_configs rKc Cs,t}t|jd}||W5QRX|S)Nr*)rr+Zconfig_file_namer,)textr!r/r r r rEsrEcCs<ddlm}tttjs(tt|tjtdS)NrConfigztest_alembic.ini) Zalembic.configrNraccessr F_OKmkdirrrrMr r r rs  rasciic Cs|j|}|j}t|}|r*||}t|d}||W5QRXt |}|rbt |t ||} |j| j}|j| jkrtd|jj| dd|rt||dkrdnddS)Nwbz2Can't change down_revision on a refresh operation.T)_replacerrr)Z revision_map get_revisionrtextwrapdedentencoder+r,rr1rr2r Z _from_pathrevisionZ down_revision ExceptionZ add_revisionr) Z scriptdirrev_idcontentencodingr oldrfpr3rr r r write_scripts,        r`cCsddl}|||dkrZt|}tjjd}tj |\}}||}t |||}n|dksft t|}t |tjst t|dS)Nrr)rr) py_compilecompilerr1 importlib machineryBYTECODE_SUFFIXESrrsplitextr'moverrOrPr2)rstylerar3suffixfilepathextZsimple_pyc_pathr r r rs      rcCst}t}t}t|}|j|ddddt||d||j|dd|dt||d|d|d d d |j|d d|dt||d ||f|||fS)Nz revision aTbase)refreshheadz"Rev A" revision = '%s' down_revision = None from alembic import op def upgrade(): op.execute("CREATE STEP 1") def downgrade(): op.execute("DROP STEP 1") z revision bu/# coding: utf-8 "Rev B, méil, %3" revision = 'z' down_revision = 'z}' from alembic import op def upgrade(): op.execute("CREATE STEP 2") def downgrade(): op.execute("DROP STEP 2") zutf-8)r]z revision cz"Rev C" revision = '%s' down_revision = '%s' from alembic import op def upgrade(): op.execute("CREATE STEP 3") def downgrade(): op.execute("DROP STEP 3") rr[r rZgenerate_revisionr`)r!abcrr r r three_rev_fixture1s@ rscCst}t}t}t|}|j|d|dddt||d||f|j|d|dddt||d||f|j|d|dddt||d||f|||fS) z:Create a multiple head fixture from the three-revs fixturezrevision d from bT)rnZsplicermz"Rev D" revision = '%s' down_revision = '%s' from alembic import op def upgrade(): op.execute("CREATE STEP 4") def downgrade(): op.execute("DROP STEP 4") zrevision e from dz"Rev E" revision = '%s' down_revision = '%s' from alembic import op def upgrade(): op.execute("CREATE STEP 5") def downgrade(): op.execute("DROP STEP 5") zrevision f from bz"Rev F" revision = '%s' down_revision = '%s' from alembic import op def upgrade(): op.execute("CREATE STEP 6") def downgrade(): op.execute("DROP STEP 6") ro)r!rprqrrder/rr r r multi_heads_fixture~s^ rvcCs`tjtd}dtjjjk}d|}ddd| D}t d||rRdnd ||fS) z?alembic.ini fixture to work exactly with the 'multidb' templaterr8z, z css |]\}}d||jfVqdS)z[%s] sqlalchemy.url = %sN)r7).0keyvaluer r r sz*_multidb_testing_config..a [alembic] script_location = %s sourceless = false sqlalchemy.future = %s databases = %s %s [loggers] keys = root [handlers] keys = console [logger_root] level = WARNING handlers = console qualname = [handler_console] class = StreamHandler args = (sys.stderr,) level = NOTSET formatter = generic [formatters] keys = generic [formatter_generic] format = %%(levelname)-5.5s [%%(name)s] %%(message)s datefmt = %%H:%%M:%%S r>r?) rrrr rrBrCrDkeysitemsrE)r%r.rFZ databasesr r r _multidb_testing_configs  r})TrF)r5FN)FF)Fr@)rIr@)rRF)Zimportlib.machineryrcrr'rVr&rrr@rr:rrr r r r#r)r0r4r<rGrHrKrErr`rrsrvr}r r r r s:         "  4 1 ( MY