B `jA @sdZddlmZmZmZeZddlZddlZddl Z ddl Z ddl Z ddl Z ddl Z ddlmZddlmZddlmZmZmZmZmZmZmZmZmZmZmZmZddl m!Z!m"Z"dd l#m$Z$dd l%m&Z&iZ'iZ(iZ)Gd d d Z*e*+Gd ddZ,GdddZ-ddZ.ddZ/ddZ0ddZ1ddZ2ddZ3ddZ4dd Z5ej6d!d"Z7d:d$d%Z8d&d'Z9d(d)Z:d;d*d+Z;d,d-Zd=d3d4Z?d5d6Z@d>d8d9ZAdS)?z1Common utility code that depends on CommonConfig.)absolute_importdivisionprint_functionN)types)to_bytes) common_environmentCOVERAGE_CONFIG_NAMEdisplay find_python remove_treeMODE_DIRECTORYMODE_FILE_EXECUTE PYTHON_PATHS raw_commandread_lines_without_commentsANSIBLE_TEST_DATA_ROOTApplicationError)write_text_filewrite_json_file) data_context)LayoutMessagesc@s`eZdZdZdZdZdZdZdZdZ dZ e ddZ ddZ eddZed d Zd d ZdS) ResultTypezTest result type.NcCsJtdt_tdt_tdt_tdt_tdt_tdt_tdt_dS)NZbotcoveragedataZjunitZlogsZreportsz.tmp)rBOTCOVERAGEDATAJUNITLOGSREPORTSTMPr"r"Q/home/dcms/DCMS/lib/python3.7/site-packages/ansible_test/_internal/util_common.py _populate>s      zResultType._populatecCs ||_dS)N)name)selfr%r"r"r#__init__HszResultType.__init__cCstjtjj|jS)z)The content relative path to the results.)ospathjoinrcontent results_pathr%)r&r"r"r# relative_pathKszResultType.relative_pathcCstjtjj|jS)z!The absolute path to the results.)r(r)r*rr+rootr-)r&r"r"r#r)PszResultType.pathcCs|jS)N)r%)r&r"r"r#__str__UszResultType.__str__)__name__ __module__ __qualname____doc__rrrrrr r! staticmethodr$r'propertyr-r)r/r"r"r"r#r4s   rc@s eZdZdZddZddZdS) CommonConfigz%Configuration common to all commands.cCsF||_|j|_|j|_|j|_|j|_|j|_|j|_d|_i|_dS)z< :type args: any :type command: str FN) commandcolorexplain verbositydebugtruncateZredactZ info_stderrcache)r&argsr7r"r"r#r'_szCommonConfig.__init__cCstjtdS)z;Return the path to the Ansible config for the given config.z ansible.cfg)r(r)r*r)r&r"r"r#get_ansible_configqszCommonConfig.get_ansible_configN)r0r1r2r3r'r?r"r"r"r#r6]sr6c@seZdZdZddZdS)NetworkPlatformSettingsz6Settings required for provisioning a network platform.cCs||_||_dS)N) collectioninventory_vars)r&rArBr"r"r#r'xsz NetworkPlatformSettings.__init__N)r0r1r2r3r'r"r"r"r#r@vsr@cCs ttdS)z+ :rtype: dict[str, dict[str, str]] Zdocker)get_parameterized_completionDOCKER_COMPLETIONr"r"r"r#get_docker_completion}srEcCs ttdS)z+ :rtype: dict[str, dict[str, str]] remote)rCREMOTE_COMPLETIONr"r"r"r#get_remote_completionsrHcCs ttdS)z+ :rtype: dict[str, dict[str, str]] network)rCNETWORK_COMPLETIONr"r"r"r#get_network_completionsrKcs^|sZtjjrdndttjtdd|dd}|t fddd d |DD|S) zj :type cache: dict[str, dict[str, str]] :type name: str :rtype: dict[str, dict[str, str]] rAz ansible-base completionz%s.txtT)Zremove_blank_linesc3s*|]"}|r|ddkr|VqdS)rcontextN)get).0kvp)rMr"r# sz/get_parameterized_completion..cSsg|] }t|qSr")parse_parameterized_completion)rOir"r"r# sz0get_parameterized_completion..) rr+rArr(r)r*rupdatedict)r=r%Zimagesr")rMr#rCs &rCcCsD|}|sdS|d}tdddd|ddDD}||fS)zbParse the given completion entry, returning the entry name and a dictionary of key/value settings.Nrcss.|]&}|dt|dkr |dndfVqdS)rrN)len)rOrPr"r"r#rQsz1parse_parameterized_completion..cSsg|]}|ddqS)=r)split)rOitemr"r"r#rTsz2parse_parameterized_completion..r)rZrV)valuevaluesr%rr"r"r#rRs $rRcCst|i}|d|S)z) :type name: str :rtype: str r%)rErN)r%configr"r"r#docker_qualify_imagesr_c Csdd||f}t|i}|j||d}t|t|j||d|rVd||fn|d}|S)zplatformversionplatform_versionrLrAsettingsr"r"r#get_network_settingss recCsZ|sdSx|jD]}tj|ddqWx|jD]}t|q.W|jrVtd|jdS)z"Display the given layout messages.Nr)r: )infor warningerrorrr*)messagesmessager"r"r#handle_layout_messagess  rlc cs^|jr tj|d||fVn:tj|||d$}|t|||j VWdQRXdS)z :param args: CommonConfig :param prefix: str :param suffix: str :param directory: str :param content: str | bytes | unicode :rtype: str z%stemp%s)prefixsuffixdirN) r9r(r)r*tempfileNamedTemporaryFilewriterflushr%)r>rmrn directoryr+Z tempfile_fdr"r"r#named_temporary_files ruTcCs&tj|j|}t||d||ddS)z`Write the given json content to the specified test results path, creating directories as needed.T)create_directories formattedencoderN)r(r)r*r)categoryr%r+rwrxr)r"r"r#write_json_test_resultssrzcCs"tj|j|}t||dddS)z`Write the given text content to the specified test results path, creating directories as needed.T)rvN)r(r)r*r)ryr%r+r)r"r"r#write_text_test_resultssr{cCst|}|r|Sd}d}d}|jr>tj|d|d|fStj|||d}tj|d}tjtj |tj|k}|rt j d||fd d t ||n t j d ||fd d t ||t|ttstt|t|<|S) zK :type args: TestConfig :type interpreter: str :rtype: str zpython-z-ansiblez/tmprWtemp)rmrnropythonz4Injecting "%s" as a symlink to the "%s" interpreter.r)r:z;Injecting "%s" as a execv wrapper for the "%s" interpreter.)rrNr9r(r)r*rpmkdtempdirnamerealpathr rgsymlinkcreate_interpreter_wrapperchmodr atexitregistercleanup_python_paths)r> interpreter python_pathrmrnZ root_temp_dirinjected_interpreterZ use_symlinkr"r"r#get_python_paths*     rcCs(tj|p d|pd|d}tt||S)zKCreate a temporary directory that persists until the current process exits.tmprW)rmrnro)rpr~rrr )rmrnbase_dir temp_pathr"r"r#create_temp_dir.s rcCs6tj}td||f}t||t|tdS)zHCreate a wrapper for the given Python interpreter at the specified path.z #!%s from __future__ import absolute_import from os import execv from sys import argv python = '%s' execv(python, [python] + argv[1:]) N) sys executabletextwrapdedentlstriprr(rr)rrZshebang_interpretercoder"r"r#r5s   rcCs6x0ttD] }tjd|ddt|qWdS)z*Clean up all temporary python directories.z*Cleaning up temporary python directory: %s)r:N)sortedrr]r rgshutilrmtree)r)r"r"r#rMsrc Cs|r|}|}n0|jr6|j}tjtjjtjj}ntdtj|t }tj|t j j d|j ||jpnd|d|f} |jstj|std||jrd} t|| d} |r | t|| d|r tj|d |j ||jpd f| d <tjtjjd | d <| S)z :type args: TestConfig :type target_name: str :type version: str :type temp_path: str :type module_coverage: bool :type remote_temp_path: str | None :rtype: dict[str, str] zXNo temp path and no coverage config base path. Check for missing coverage_context usage.z%s=%s=%s=%s=coveragezlocal-%sz python-%sz Missing coverage config file: %srW)Z COVERAGE_CONFZ COVERAGE_FILE)Z_ANSIBLE_COVERAGE_CONFIGZ_ANSIBLE_COVERAGE_OUTPUTz%s=%s=%srFZ_ANSIBLE_COVERAGE_REMOTE_OUTPUT*Z"_ANSIBLE_COVERAGE_REMOTE_WHITELIST)coverage_config_base_pathr(r)r*rr+r.r, Exceptionr rrr%r7Zcoverage_labelr9existsZcoverage_checkrVrU) r> target_namerbrmodule_coverageremote_temp_pathrZcoverage_output_base_path config_fileZ coverage_fileenvr"r"r#get_coverage_environmentTs4     rFc  Cs|s t}n|}t|}|p$|j} | p0t| }tjtd}| s^t ||}|tjj |}|tjj |d|d<| |d<||d<|j r| s| t ||| || | dt||||||dS)a :type args: TestConfig :type cmd: collections.Iterable[str] :type target_name: str :type env: dict[str, str] :type capture: bool :type data: str | None :type cwd: str | None :type python_version: str | None :type temp_path: str | None :type module_coverage: bool :type virtualenv: str | None :type disable_coverage: bool :type remote_temp_path: str | None :rtype: str | None, str | None ZinjectorPATHZANSIBLE_TEST_PYTHON_VERSIONZANSIBLE_TEST_PYTHON_INTERPRETER)r)capturerrcwd)rcopylistpython_versionr r(r)r*rrpathseprrUr run_command)r>cmdrrrrrrrr virtualenvZdisable_coveragerrbrZ inject_pathrr"r"r#intercept_commands"     rcCsttjjs|ds|S|d}|tjj}x|D]}|rB|rDP|d=q6Wdtjjd |dd|DfS)z4 :type import_name: str :type path: str .zansible_collections.%s%scSsg|] }|r|qSr"r")rOpr"r"r#rTsz*resolve_csharp_ps_util..) rr+Z is_ansible startswithrZr(r)seprmr*) import_namer)packagesZmodule_packagespackager"r"r#resolve_csharp_ps_utils    rstrictc Cs(|jo | } t|||||| ||| | d S)ai :type args: CommonConfig :type cmd: collections.Iterable[str] :type capture: bool :type env: dict[str, str] | None :type data: str | None :type cwd: str | None :type always: bool :type stdin: file | None :type stdout: file | None :type cmd_verbosity: int :type str_errors: str :rtype: str | None, str | None ) rrrrr9stdinstdout cmd_verbosity str_errors)r9r) r>rrrrralwaysrrrrr9r"r"r#rs r)TN)NNN)N) FNNNNTNFN) FNNNFNNrr)Br3 __future__rrrtype __metaclass__r contextlibr(rrrprrWrtencodingrutilrr r r r r rrrrrriorrrrZprovider.layoutrrDrGrJrr$r6r@rErHrKrCrRr_rerlcontextmanagerrurzr{rrrrrrrrr"r"r"r#sX  8  &   -  > -