/usr/lib/python3/dist-packages/landscape/lib/__pycache__
Edit: /usr/lib/python3/dist-packages/landscape/lib/__pycache__/testing.cpython-310.pyc (26194B)
o
cV`
@ sH d dl mZ d dlZd dlZd dlZd dlZd dlZd dlZd dlZd dl Z d dl
Z
d dl Z d dlm Z m
Z
mZ d dlmZ d dlmZ d dlmZ d dlmZ d dlmZ d d lmZ d d
lm Z d d lm Z m Z d d l m Z d d
l!m"Z" d dl#m$Z$ G dd de jZ%G dd de jZ&G dd de'Z(G dd de(Z)G dd deZ*G dd de Z+G dd d e,Z-G d d d e'Z.G d d d e'Z/G d!d" d"e'Z0G d#d$ d$e'Z1G d%d& d&e'Z2d d d'd(d(d(d d d d d g d)f d*d+Z3d:d,d-Z4d.d/ Z5G d0d1 d1e'Z6G d2d3 d3e'Z7G d4d5 d5e'Z8G d6d7 d7e'Z9G d8d9 d9e"Z:dS ); )absolute_importN)HandlerERROR Formatter)TestCase)
StringType)_PY3)Failure)Deferred) ConnectError) ConfigParser)stringio cstringio)BaseConfiguration)EventHandlingReactorMixin) LoginInfoc @ s e Zd Zes ejZdS dS )CompatTestCaseN)__name__
__module__ __qualname__r r assertItemsEqualassertCountEqual r r 7/usr/lib/python3/dist-packages/landscape/lib/testing.pyr s
r c s, e Zd Zg Z fddZ fddZ ZS )HelperTestCasec sf t t| g | _t| jvr| jdt d }| jD ]}| }t|dr*|| }| j | q|S )Nr set_up)
superr setUp_helper_instancesLogKeeperHelperhelpersinserthasattrr append)selfresulthelper_factoryhelper __class__r r r ( s
zHelperTestCase.setUpc s6 t | jD ] }t|dr|| qtt| d S )N tear_down)reversedr r" r* r r tearDown)r$ r' r( r r r, 8 s
zHelperTestCase.tearDown)r r r r r r,
__classcell__r r r( r r $ s r c @ s> e Zd Zdd Z dddZd d
Zdd d Zdd
dZdS )
FSTestCasec C sB t |d }| }W d n1 sw Y | || d S )Nrb)openread assertEqual)r$ filenameexpected_contentfdactual_contentr r r assertFileContentB s
z FSTestCase.assertFileContentN tmpwc s |dur|du r t }tj||ndu r/t |||\} t| |du r/t |durMt|
}
|
| W d n1 sHw Y |
| j rb fdd} |
| S )a Create a temporary file and return the path to it.
@param content: Initial content for the file.
@param suffix: Suffix to be given to the file's basename.
@param prefix: Prefix to be given to the file's basename.
@param basename: Full basename for the file.
@param dirname: Put file inside this directory.
The file is removed after the test runs.
Nc s* z
t W d S ty Y d S w N)osremoveOSErrorr backupsuffixpathr r
remove_backupc s
z*FSTestCase.makeFile.
.remove_backup) tempfilemkdtempr< rA joinmkstempcloseunlinkr0 write
addCleanup _clean_file) r$ contentsuffixprefixbasenamedirnamerA moder@ r5 filerB r r? r makeFileG s$
zFSTestCase.makeFilec C sB zt j|rt| W dS t | W dS ty Y dS w )ztTry to remove a filesystem path, whether it's a directory or file.
@param path: the path to remove
N)r< rA isdirshutilrmtreerH r> )r$ rA r r r rK l s zFSTestCase._clean_filec C s4 |dur
t | nt|||}| | j| |S )a@ Create a temporary directory and return the path to it.
@param suffix: Suffix to be given to the file's basename.
@param prefix: Prefix to be given to the file's basename.
@param dirname: Put directory inside this parent directory.
The directory is removed after the test runs.
N)r< makedirsrC rD rJ rK )r$ rM rN rP rA r r r makeDiry s
zFSTestCase.makeDirc C s6 |du r| }||_| j|||d}t|d |S )z=Return the path to the script after writing it to a temp dir.N)rP rO i )rX bindirrS r< chmod)r$ confignamerL rY r3 r r r write_script s zFSTestCase.write_script)Nr8 r9 NNNr: N)r8 r9 NNr; )r r r r7 rS rK rX r] r r r r r. @ s
%
r. c s0 e Zd Z fddZ fddZdd Z ZS )ConfigTestCasec s( t t| tj| _| dgt_d S Nr8 )r r^ r r default_config_filenames_old_config_filenamesrS r$ r( r r r s zConfigTestCase.setUpc s | j t_tt| d S r; )ra r r` r r^ r, rb r( r r r, s zConfigTestCase.tearDownc C s~ t |}t }|| t |}t }|| | t| t| | D ]}| t||t|| q*dS )z
Compare two configuration files for equality. The order of parameters
and comments may be different but the actual parameters and sections
must be the same.
N)r r readfpr2 setsectionsdictitems)r$ firstsecondfirst_fp first_parser second_fp
second_parsersectionr r r assertConfigEqual s
z ConfigTestCase.assertConfigEqual)r r r r r, ro r- r r r( r r^ s r^ c @ s6 e Zd Zdd Zdd Zdd Zdd Zd d
d Zd S )
TwistedTestCasec C sX g }| |j |s| d|f dS t|d tr(| d||d f dS |d S )zSee C{twisted.trial._synctest._Assertions.successResultOf}.
This is a copy of the original method, which is available only
since Twisted 12.3.0 (from 2012-12-20).
z6Success result expected on %r, found no result insteadr z@Success result expected on %r, found failure result (%r) insteadNaddBothr# fail
isinstancer r$ deferredr% r r r successResultOf
z TwistedTestCase.successResultOfc C sX g }| |j |s| d|f dS t|d ts(| d||d f dS |d S )zSee C{twisted.trial._synctest._Assertions.failureResultOf}.
This is a copy of the original method, which is available only
since Twisted 12.3.0 (from 2012-12-20).
z6Failure result expected on %r, found no result insteadr z@Failure result expected on %r, found success result (%r) insteadNrq ru r r r failureResultOf rx z TwistedTestCase.failureResultOfc C s2 g }| |j |r| d||d f dS dS )zSee C{twisted.trial._synctest._Assertions.assertNoResult}.
This is a copy of the original method, which is available only
since Twisted 12.3.0 (from 2012-12-20).
z*No result expected on %r, found %r insteadr N)rr r# rs ru r r r assertNoResult s z TwistedTestCase.assertNoResultc s8 | t|t g fdd}|| | d S )Nc s d d S NT)r# )r% calledr r callback z9TwistedTestCase.assertDeferredSucceeded..callback)
assertTruert r
addCallback)r$ rv r~ r r| r assertDeferredSucceeded s
z'TwistedTestCase.assertDeferredSucceededNc C s | t|t || j|S )zS
Assert that the given C{deferred} results in the given C{result}.
)r rt r
r r2 ru r r r
assertSuccess s z TwistedTestCase.assertSuccessr; )r r r rw ry rz r r r r r r rp s
rp c @ e Zd Zdd Zdd ZdS ) ErrorHandlerc O s" t j| g|R i | g | _d S r; )r __init__errors)r$ argskwargsr r r r s
zErrorHandler.__init__c C s |j tkr
| j| d S d S r; )levelnor r r# )r$ recordr r r emit s
zErrorHandler.emitN)r r r r r r r r r r s r c @ e Zd Zdd ZdS )LoggedErrorsErrorc C s0 d}t }| jd D ] }|||d 7 }q
|S )Nz!The following errors were logged
r
)r r format)r$ out formattererrorr r r __str__ s
zLoggedErrorsError.__str__N)r r r r r r r r r r c @ ( e Zd ZdZdd Zdd Zdd ZdS ) r zRecord logging information.
Puts a 'logfile' attribute on your test case, which is a StringIO
containing all log output.
c C s| g | _ g | _t | _| |_t |_}t |_ t
|j }d}| t | |j
| _|j| _|| jg|_
|tj d S )Nz%(levelname)8s: %(message)s)ignored_exception_regexesignored_exception_typesr
error_handler
log_helperlogging getLoggerloggerr logfile
StreamHandler setFormatterr handlers old_handlerslevel old_levelsetLevelNOTSET)r$ test_caser handlerr r r r r s zLogKeeperHelper.set_upc C s t }|| j | j|_g }| jjD ]-}| jD ]}|j r-|j d r-t
|j d |r- nq| j D ]
}| |j
r; nq1|| q|rHt|d S Nr )r r r r r r r r r exc_info
issubclassr matchmessager# r )r$ r r r r ignored_type
ignored_regexr r r r* ) s(
zLogKeeperHelper.tear_downc C s0 t |tr| jt| d S | j| d S r; )rt
basestringr r# recompiler )r$
type_or_regexr r r
ignore_errors= s
z LogKeeperHelper.ignore_errorsN)r r r __doc__r r* r r r r r r s
r c @ r )EnvironSnapshotc C s t j | _d S r; )r< environcopy _snapshotrb r r r r F zEnvironSnapshot.__init__c C s4 t j| j tt jD ] }|| jvrt j|= q d S r; )r< r updater list)r$ keyr r r restoreI s
zEnvironSnapshot.restoreN)r r r r r r r r r r D r c @ r )EnvironSaverHelperc C s t | _d S r; )r r r$ r r r r r R s zEnvironSaverHelper.set_upc C s | j d S r; )r r r r r r r* U r z EnvironSaverHelper.tear_downNr r r r r* r r r r r P r r c @ sH e Zd ZdddZdddZdddZd d
Zdd d Zed
d Z dS ) MockPopenNr8 c C s, || _ || _t|| _g | _|| _d | _d S r; )outputerr_outr stdout popen_inputs return_codesreceived_input)r$ r r r r r r r [ s
zMockPopen.__init__c C s | j ||||dS )N)stdinr stderr)popenr$ r r r r r r r __call__c s zMockPopen.__call__c C s | j | | S r; )r r# r r r r r f s zMockPopen.popenc C s | j S r; )
returncoderb r r r waitj s zMockPopen.waitc C s || _ | j| jfS r; )r r r )r$ inputr r r communicatem s zMockPopen.communicatec C s | j d u rdS | j dS r )r poprb r r r r q s
zMockPopen.returncoder_ )NNNr; )
r r r r r r r r propertyr r r r r r Y s
r c @ r )StandardIOHelperc C s@ t j|_t j|_t |_t _t |_t _ts d|j_d S d S )NzUTF-8)sysr
old_stdoutr old_stdinr
r encodingr r r r r z s zStandardIOHelper.set_upc C s |j t_|jt_d S r; )r r r r r r r r r r* s zStandardIOHelper.tear_downNr r r r r r x s r z/dev/r8 )r r r r c
C s~ t | d}
z4|
ttj|||d|d|d|d||| |
| | d | d | d | d d W |
dS |
w ) z3Append binary login data to the specified filename.abzutf-8r N)r0 rI structpackr
RAW_FORMATencoderG )r3
login_typepid
tty_deviceidusernamehostnametermination_status exit_status
session_identry_time_secondsentry_time_millisecondsremote_ip_addressrR r r r append_login_data s
r c c s | V | d7 } q)z>Generator starts at zero and yields integers that grow by one.Tr r )ir r r mock_counter s
r c C s t dS )zCGenerator starts at 100 and yields int timestamps that grow by one.d )r r r r r mock_time s r c @ s0 e Zd ZdZdd Zdi dddddfddZdS ) StubProcessFactoryzx
A L{IReactorProcess} provider which records L{spawnProcess} calls and
allows tests to get at the protocol.
c C
g | _ d S r; )spawnsrb r r r r
zStubProcessFactory.__init__r Nr c
C s" | j ||||||||| f d S r; )r r# )
r$ protocol
executabler envrA uidgidusePTYchildFDsr r r spawnProcess s
z StubProcessFactory.spawnProcess)r r r r r r r r r r r s r c @ r ) DummyProcessz/A process (transport) that doesn't do anything.c C r r; )signalsrb r r r r r zDummyProcess.__init__c C s | j | d S r; )r r# )r$ signalr r r
signalProcess r zDummyProcess.signalProcessc C s d S r; r )r$ r5 r r r closeChildFD s zDummyProcess.closeChildFDN)r r r r r r r r r r r r s
r c @ sJ e Zd ZdZdZdZdZdZdZdZ dZ
d d
Z dddZ dd Z
d S )ProcessDataBuilderzBuilder creates sample data for the process info plugin to consume.
@param sample_dir: The directory for sample data.
z R (running)z T (stopped)zT (tracing stop)zD (disk sleep)z S (sleeping)zX (dead)z
Z (zombie)c C s
|| _ d S r; ) _sample_dir)r$
sample_dirr r r r r zProcessDataBuilder.__init__r NT- c
C s d|dd |||| d }
t j| jt|} t | t j| d} t| d}
z |
|
W |
n|
w |du rCd|f }t j| d} t| d}
z |
| W |
n|
w |rid d
|i }
nd }
t j| d } t| d}
z |
|
W |
dS |
w )
a Creates sample data for a process.
@param started_after_boot: The amount of time, in jiffies,
between the system uptime and start of the process.
@param process_name: Used to generate the process name that appears in
/proc/%(pid)s/status
@param generate_cmd_line: If true, place the process_name in
/proc/%(pid)s/cmdline, otherwise leave it empty (this simulates a
kernel process)
@param stat_data: Array of items to write to the /proc//stat file.
a
Name: %(process_name)s
State: %(state)s
Tgid: 24759
Pid: 24759
PPid: 17238
TracerPid: 0
Uid: %(uid)d 0 0 0
Gid: %(gid)d 0 0 0
FDSize: 256
Groups: 4 20 24 25 29 30 44 46 106 110 112 1000
VmPeak: 11680 kB
VmSize: %(vmsize)d kB
VmLck: 0 kB
VmHWM: 6928 kB
VmRSS: 6924 kB
VmData: 1636 kB
VmStk: 196 kB
VmExe: 1332 kB
VmLib: 4240 kB
VmPTE: 20 kB
Threads: 1
SigQ: 0/4294967295
SigPnd: 0000000000000000
ShdPnd: 0000000000000000
SigBlk: 0000000000000000
SigIgn: 0000000000000000
SigCgt: 0000000059816eff
CapInh: 0000000000000000
CapPrm: 0000000000000000
CapEff: 0000000000000000
N ) process_namestater r vmsizestatuszw+z,0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 %dstatzE/usr/sbin/%(process_name)s --pid-file /var/run/%(process_name)s.pid
r
r8 cmdline) r< rA rE r strmkdirr0 rI rG )r$
process_idr r r started_after_bootr
generate_cmd_line stat_datar sample_data process_dirr3 rR r r r create_data s> !
z ProcessDataBuilder.create_datac C s" t j| jt|}t| dS )z>Remove sample data for the process that matches C{process_id}.N)r< rA rE r r rU rV )r$ r r r r r remove_data( s z ProcessDataBuilder.remove_data)r NTNr )r r r r RUNNINGSTOPPED TRACING_STOP
DISK_SLEEPSLEEPINGDEADZOMBIEr r r r r r r r s
Sr c @ r )
FakeReactorIDc C s d| _ || _d S r{ )active_data)r$ datar r r r 0 s
zFakeReactorID.__init__N)r r r r r r r r r! . r r! c s e Zd ZdZi Z fddZdd Zdd Zdd Zd
d Z fd d
Z
dd Z dd Z dd Z
dd Zdd Zdd Zdd Zd d Zd d Zd d! Zd"d# Zd$d% Z ZS )& FakeReactoraw A fake reactor with the same API of L{LandscapeReactor}.
This reactor emulates the asychronous interface of L{LandscapeReactor}, but
implementing it in a synchronous way, for easier unit-testing.
Note that the C{listen_unix} method is *not* emulated, but rather inherited
blindly from L{UnixReactorMixin}, this means that there's no way to control
it in a synchronous way (see the docstring of the mixin). A better approach
would be to fake the AMP transport (i.e. fake the twisted abstractions
around Unix sockets), and implement a fake version C{listen_unix}, but this
hasn't been done yet.
c s< t t| d| _g | _i | _g | _ddlm} || _ d S )Nr )reactor)
r r% r
_current_time_callshosts_threaded_callbackstwisted.internetr& _reactor)r$ r& r( r r r G s
zFakeReactor.__init__c C s
t | jS r; )floatr' rb r r r timeS r zFakeReactor.timec O s( | j | }||||f}| | t|S r; )r' _insort_callr! )r$ secondsfr r scheduled_timecallr r r
call_laterV s
zFakeReactor.call_laterc C s2 dd | j D }t||d }| j || d S )Nc S s g | ]}|d qS r r ).0cr r r
` s z,FakeReactor._insort_call..r )r( bisect bisect_leftr! )r$ r3 timesindexr r r r/ \ s zFakeReactor._insort_callc s( fdd S )Nc sF j_z
i W d S ty" jr! w r; )r4 r# Exceptionr" cancel_callr r r3 r1 faker r0 r$ r r r@ f s
z$FakeReactor.call_every..faker4 )r$ r0 r1 r r r r? r
call_everyd s zFakeReactor.call_everyc sD t |tu r|j| jv r| j|j d|_d S tt| | d S NF) typer! r# r( r= r" r r% r> r$ r r( r r r> t s
zFakeReactor.cancel_callc C s | d| d S r rA )r$ r1 r r r call_when_running| s z FakeReactor.call_when_runningc s | j fdd dS )z5Schedule a function for execution in the main thread.c s i S r; r r r r1 r r r s z*FakeReactor.call_in_main..N)r* r# )r$ r1 r r r rG r call_in_main s zFakeReactor.call_in_mainc O s | ||||| | dS )as Emulate L{LandscapeReactor.call_in_thread} without spawning threads.
Note that running threaded callbacks here doesn't reflect reality,
since they're usually run while the main reactor loop is active. At
the same time, this is convenient as it means we don't need to run
the the real Twisted reactor with to test actions performed on
completion of specific events (e.g. L{MessageExchange.exchange} uses
call_in_thread to run the HTTP request in a separate thread, because
we use libcurl which is blocking). IOW, it's easier to test things
synchronously.
N)
_in_thread_run_threaded_callbacks)r$ r~ errbackr1 r r r r r call_in_thread s zFakeReactor.call_in_threadc s& G fdddt }| j< | S )Nc s e Zd Z fddZdS )z)FakeReactor.listen_unix..FakePortc s j d S r; )
_socket_pathsr )oselfr$ socket_pathr r
stopListening r z7FakeReactor.listen_unix..FakePort.stopListeningN)r r r rR r rP r r FakePort s rS )objectrN )r$ rQ factoryrS r rP r listen_unix s
zFakeReactor.listen_unixc C sT | j |}ddlm} |r|||}| |S t }ttd}||| |S )Nr )
FakeConnectorzNo such file or directory) rN get landscape.lib.tests.test_amprW connectrT r r clientConnectionFailed)r$ rA rU serverrW connectorfailurer r r connect_unix s
zFakeReactor.connect_unixc C s> | d d| _| jr| | jd d | js | d dS )zAContinuously advance this reactor until reactor.stop() is called.runTr stopN)fire_runningadvancer( rb r r r r` s
zFakeReactor.runc C s
d| _ d S rC )rc rb r r r ra r zFakeReactor.stopc
C s | j rY| j d d | j| krY| j d}||d | j 8 }|d | _z|d |d i |d W n tyI } z t| W Y d}~nd}~ww | j rY| j d d | j| ks| j|7 _dS )aH Advance this reactor C{seconds} into the future.
This method is not part of the L{LandscapeReactor} API and is specific
to L{FakeReactor}. It's meant to be used only in unit tests for
advancing time and triggering the relevant scheduled calls (see
also C{call_later} and C{call_every}).
r r r r N)r( r' r r= r exception)r$ r0 r3 er r r rd s
zFakeReactor.advancec
C s z ||i |}W n5 t y> } z)t }|d u r#| jtj||d n| j|g|R W Y d }~d S W Y d }~d S d }~ww |rI| || d S d S )N)r )r= r r rI r r ) r$ r~ rL r1 r r r% rf r r r r rJ s zFakeReactor._in_threadc
C sV | j r)z | j d W n ty# } z t| W Y d }~nd }~ww | j sd S d S r )r* r r= r re )r$ rf r r r rK s z#FakeReactor._run_threaded_callbacksc C s | d| j}|| _d S )Ng ?)rB rK _run_threaded_callbacks_idrE r r r _hook_threaded_callbacks s
z$FakeReactor._hook_threaded_callbacksc C s | | j d S r; )r> rg rb r r r _unhook_threaded_callbacks r z&FakeReactor._unhook_threaded_callbacks)r r r r rN r r. r4 r/ rB r> rF rI rM rV r_ r` ra rd rJ rK rh ri r- r r r( r r% 5 s*
r% r5 );
__future__r r9 r r< os.pathr rU r r rC unittestr r r twisted.trial.unittestr twisted.python.compatr r landscape.lib.compatr twisted.python.failurer twisted.internet.deferr
twisted.internet.errorr r r
r landscape.lib.configr landscape.lib.reactorr landscape.lib.sysstatsr r r rT r. r^ rp r r= r r r r r r r r r r r r r! r% r r r r s` V"F
1
j