B üÏf*ã@sddlZddlZddlZddlZddlmZddlmZddlm Z ddlm Z ddlm Z ddlm Z dd l m Z dd l mZd d „Zd2dd„Zdd„Zdd„Zdd„Zd3dd„Zd4dd„Zd5dd„Zd6d!d"„Zd#d$„Zd%d&„Zd7d(d)„Zd*d+„Zd,d-„Zd.d/„Zd0d1„ZdS)8éN)Úconfig)Ú provisioné)Úutilé)Úcommand)Úscript)ÚScript)ÚScriptDirectorycCstjrdtjSdSdS)Nz scratch_%sZscratch)rZFOLLOWER_IDENT©r r úUC:\Users\Lenovo\AppData\Local\Temp\pip-unpacked-wheel-q6xnih48\alembic\testing\env.pyÚ_get_staging_directorys r TÚgenericFcCs¬tƒ}|rœtj tƒd¡}tj |¡r2td|ƒ‚tj|||d|rœyt   |d¡Wnt k rjYnX|dks|t|ƒ‚t tj |d¡d|kr–dndƒt j |¡}|S)NÚscriptsz2staging directory %s already exists; poor cleanup?)Útemplatezenv.py)Zpep3147_envonlyÚsimpleÚpep3147_everythingÚpep3147r)Ú_testing_configÚosÚpathÚjoinr ÚexistsÚAssertionErrorrÚinitrZload_python_fileÚAttributeErrorÚmake_sourcelessrr Ú from_config)ÚcreaterÚ sourcelessÚcfgrZscr r r Ú staging_envs&   r!cCs(ddlm}|j ¡t tƒd¡dS)Nr)ÚenginesT)Úsqlalchemy.testingr"Ztesting_reaperZ close_allÚshutilÚrmtreer )r"r r r Úclear_staging_env:s  r&c CsBtj tƒd¡}tj |d¡}t|dƒ}| |¡WdQRXdS)Nrzscript.py.makoÚw)rrrr ÚopenÚwrite)ÚtxtÚdir_rÚfr r r Úscript_file_fixtureAs r-c Csbtj tƒd¡}d|}tj |d¡}t |¡}|r>t |¡t|dƒ}| |¡WdQRXdS)Nrz6 from alembic import context config = context.config zenv.pyr') rrrr rÚpyc_file_from_pathÚunlinkr(r))r*r+rÚpyc_pathr,r r r Úenv_file_fixtureHs   r1úfoo.dbcKs>tj tƒd¡}d||f}|r.tjr.||d<tj|||dS)Nrzsqlite:///%s/%sÚscope)ÚurlÚfutureÚoptions)rrrr rZsqla_14Ú testing_utilZtesting_engine)Útempnamer5r3r6r+r4r r r Ú_sqlite_file_db\s   r9cCsNtj tƒd¡}d|}|p(dtjjjk}td|||r:dnd|rDdndfƒS) Nrzsqlite:///%s/foo.dbr5aë [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 ÚtrueÚfalsezsqlalchemy.future = trueÚ) rrrr rÚdbÚ __class__Ú __module__Ú_write_config_file)rr5r+r4Úsqlalchemy_futurer r r Ú_sqlite_testing_configds$ rBr<cCsLtj tƒd¡}dtjjjk}d|}td|||r6dnd|r@dnd|fƒS)Nrr5zsqlite:///%s/foo.dbaö [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 r:r;) rrrr rr=r>r?r@)rZextra_version_locationr+rAr4r r r Ú_multi_dir_testing_config˜s   rCÚ postgresqlcCs"tj tƒd¡}td|||fƒS)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 )rrrr r@)ÚdialectZ directivesr+r r r Ú_no_sql_testing_configÉsrFc Cs,tƒ}t|jdƒ}| |¡WdQRX|S)Nr')rr(Zconfig_file_namer))Útextr r,r r r r@ñsr@cCs<ddlm}t tƒtj¡s(t tƒ¡|tj tƒd¡ƒS)Nr)ÚConfigztest_alembic.ini) Zalembic.configrHrÚaccessr ÚF_OKÚmkdirrr)rHr r r røs  rÚasciic Cs¾|j |¡}|j}t |¡}|r*| |¡}t|dƒ}| |¡WdQRXt  |¡}|rbt   |¡t   ||¡} |j | j¡}|j| jkrtdƒ‚|jj| dd|rºt||dkr´dndƒdS)NÚwbz2Can't change down_revision on a refresh operation.T)Ú_replacerrr)Z revision_mapÚ get_revisionrÚtextwrapÚdedentÚencoder(r)rr.rr/r Z _from_pathÚrevisionZ down_revisionÚ ExceptionZ add_revisionr) Z scriptdirÚrev_idÚcontentÚencodingrÚoldrÚfpr0rr r r Ú write_scripts&        rZcCsddl}| |¡|dkrZt |¡}tjjd}tj  |¡\}}||}t   ||¡|}n|dksft ‚t |¡}t  |tj¡s‚t ‚t |¡dS)Nrr)rr)Ú py_compileÚcompilerr.Ú importlibÚ machineryÚBYTECODE_SUFFIXESrrÚsplitextr$ÚmoverrIrJr/)rÚstyler[r0ÚsuffixÚfilepathÚextZsimple_pyc_pathr r r rs      rcCs¤t ¡}t ¡}t ¡}t |¡}|j|ddddt||d|ƒ|j|dd|dt||d|›d|›d d d |j|d d|dt||d ||fƒ|||fS)Nz revision aTÚbase)ÚrefreshÚheadz¨"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)rWz 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") )rrUr rÚgenerate_revisionrZ)r ÚaÚbÚcrr r r Úthree_rev_fixture1s.  rmcCs¤t ¡}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)rhZsplicergz¨"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") )rrUr rrirZ)r rjrkrlÚdÚer,rr r r Úmulti_heads_fixture~s4    rpcCs`tj tƒd¡}dtjjjk}d | ¡¡}d dd„|  ¡Dƒ¡}t d||rRdnd ||fƒS) z?alembic.ini fixture to work exactly with the 'multidb' templaterr5z, z css |]\}}d||jfVqdS)z[%s] sqlalchemy.url = %sN)r4)Ú.0ÚkeyÚvaluer 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 = 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 r:r;) rrrr rr=r>r?ÚkeysÚitemsr@)r"r+rAZ databasesr r r Ú_multidb_testing_config×s rw)TrF)r2FN)FF)Fr<)rDr<)rLF)Úimportlib.machineryr]rr$rPr#rrr<rr7rrr r r r!r&r-r1r9rBrCrFr@rrZrrmrprwr r r r Ús6         "  4 1 ( MY