o Df*@sddlZddlZddlZddlZddlmZddlmZddlm Z ddlm Z ddlm Z ddlm Z dd l m Z dd l mZd d Zd2ddZddZddZddZd3ddZd4ddZd5ddZd6d!d"Zd#d$Zd%d&Z d7d(d)Zd*d+Zd,d-Zd.d/Zd0d1ZdS)8N)config) provision)util)command)script)Script)ScriptDirectorycCstjrdtjSdS)Nz scratch_%sscratch)rFOLLOWER_IDENTr r L/home/ubuntu/webapp/venv/lib/python3.10/site-packages/alembic/testing/env.py_get_staging_directorys rTgenericFcCst}|rMtjtd}tj|rJd|tj|||d|rMzt |dWn t y4Ynw|dvs=J|t tj|dd|vrJdndt j |}|S)Nscriptsz2staging directory %s already exists; poor cleanup?)templateenv.py)pep3147_envonlysimplepep3147_everythingpep3147r)_testing_configospathjoinrexistsrinitrload_python_fileAttributeErrormake_sourcelessrr from_config)creater sourcelesscfgrscr r r staging_envs,    r&cCs(ddlm}|jttddS)NrenginesT)sqlalchemy.testingr(testing_reaper close_allshutilrmtreerr'r r rclear_staging_env:s  r.cCsXtjtd}tj|d}t|d}||WddS1s%wYdS)Nrzscript.py.makow)rrrropenwrite)txtdir_rfr r rscript_file_fixtureAs   "r5cCsxtjtd}d|}tj|d}t|}|rt|t|d}||WddS1s5wYdS)Nrz6 from alembic import context config = context.config rr/) rrrrrpyc_file_from_pathunlinkr0r1)r2r3rpyc_pathr4r r renv_file_fixtureHs    "r9foo.dbcKs>tjtd}d||f}|rtjr||d<tj|||dS)Nrzsqlite:///%s/%sscope)urlfutureoptions)rrrrrsqla_14 testing_utiltesting_engine)tempnamer=r;r>r3r<r r r_sqlite_file_db\s   rCcCsTtjtd}d|}|pdtjjjv}td|||rdnd|r%dfSdfS) Nrsqlite:///%s/foo.dbr=a [alembic] script_location = %s sqlalchemy.url = %s sourceless = %s %s [loggers] keys = root,sqlalchemy [handlers] keys = console [logger_root] level = WARN 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 rrrrrdb __class__ __module___write_config_file)r#r=r3r<sqlalchemy_futurer r r_sqlite_testing_configds % )rNrGcCsLtjtd}dtjjjv}d|}td|||rdnd|r dnd|fS)Nrr=rDa [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 = WARN 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 rErFrH)r#extra_version_locationr3rMr<r r r_multi_dir_testing_configs!  rP 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 = WARN 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 )rrrrrL)dialect directivesr3r r r_no_sql_testing_configs rTcCsBt}t|jd}||Wd|S1swY|S)Nr/)rr0config_file_namer1)textr$r4r r rrLs  rLcCs<ddlm}tttjstt|tjtdS)NrConfigztest_alembic.ini) alembic.configrXraccessrF_OKmkdirrrrWr r rrs  rasciic Cs|j|}|j}t|}|r||}t|d }||Wdn1s*wYt |}|r;t |t ||} |j| j}|j| jkrRtd|jj| dd|rit||dkrddnddSdS)Nwbz2Can't change down_revision on a refresh operation.T)_replacerrr) revision_map get_revisionrtextwrapdedentencoder0r1rr6rr7r _from_pathrevision down_revision Exception add_revisionr ) scriptdirrev_idcontentencodingr#oldrfpr8rr r r write_scripts.         rpcCsddl}|||dkr-t|}tjjd}tj |\}}||}t |||}n |dvs3Jt|}t |tj sAJt|dS)Nrr)rr) py_compilecompilerr6 importlib machineryBYTECODE_SUFFIXESrrsplitextr,moverZr[r7)rstylerqr8suffixfilepathextsimple_pyc_pathr r rr s      r cCst}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)rmz 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") rrkr r!generate_revisionrp)r$abcrr r rthree_rev_fixture1s@  rcCst}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)rsplicer~z"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") r)r$rrrder4rr r rmulti_heads_fixture~sF     rcCs`tjtd}dtjjjv}d|}ddd| D}t d||r)dnd ||fS) z?alembic.ini fixture to work exactly with the 'multidb' templaterr=z, z css"|] \}}d||jfVqdS)z[%s] sqlalchemy.url = %sN)r<).0keyvaluer r r s  z*_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 = WARN 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 rErF) rrrrrrIrJrKkeysitemsrL)r(r3rM databasesr r r_multidb_testing_configs  r)TrF)r:FN)FF)FrG)rQrG)r]F)importlib.machineryrsrr,rbr)rrrGrr@rrr r rr&r.r5r9rCrNrPrTrLrrpr rrrr r r rs:         "   4 1( M Y