synapse.tests package

Submodules

synapse.tests.nopmod module

A python module used for testing dmon dynamic module loading.

synapse.tests.utils module

This contains the core test helper code used in Synapse.

This gives the opportunity for third-party users of Synapse to test their code using some of the same helpers used to test Synapse.

The core class, synapse.tests.utils.SynTest is a subclass of unittest.TestCase, with several wrapper functions to allow for easier calls to assert* functions, with less typing. There are also Synapse specific helpers, to load Cortexes and whole both multi-component environments into memory.

Since SynTest is built from unittest.TestCase, the use of SynTest is compatible with the unittest, nose and pytest frameworks. This does not lock users into a particular test framework; while at the same time allowing base use to be invoked via the built-in Unittest library, with one important exception: due to an unfortunate design approach, you cannot use the unittest module command line to run a single async unit test. pytest works fine though.

class synapse.tests.utils.AsyncStreamEvent(*args, **kwargs)[source]

Bases: _io.StringIO, asyncio.locks.Event

A combination of a io.StringIO object and an asyncio.Event object.

setMesg(mesg)[source]

Clear the internal event and set a new message that is used to set the event.

Parameters

mesg (str) – The string to monitor for.

Returns

None

async wait(timeout=None)[source]

Block until the internal flag is true.

If the internal flag is true on entry, return True immediately. Otherwise, block until another coroutine calls set() to set the flag to true, then return True.

write(s)[source]

Write string to file.

Returns the number of characters written, which is always equal to the length of the string.

class synapse.tests.utils.CmdGenerator(cmds)[source]

Bases: object

addCmd(cmd)[source]

Add a command to the end of the list of commands returned by the CmdGenerator.

Parameters

cmd (str) – Command to add to the list of commands to return.

class synapse.tests.utils.DeprModule(core, conf=None)[source]

Bases: synapse.lib.module.CoreModule

getModelDefs()[source]
class synapse.tests.utils.HttpReflector(application: tornado.web.Application, request: tornado.httputil.HTTPServerRequest, **kwargs: Any)[source]

Bases: synapse.lib.httpapi.Handler

Test handler which reflects get/post data back to the caller

async get()[source]
async head()[source]
async post()[source]
class synapse.tests.utils.LibTst(runt, name=())[source]

Bases: synapse.lib.stormtypes.Lib

LibTst for testing!

addLibFuncs()[source]
async beep(valu)[source]

Example storm func

class synapse.tests.utils.StormPkgTest(*args, **kwargs)[source]

Bases: synapse.tests.utils.SynTest

assetdir = None
getTestCore(conf=None, dirn=None)[source]

Get a simple test Cortex as an async context manager.

Returns

A Cortex object.

Return type

s_cortex.Cortex

async initTestCore(core)[source]
pkgprotos = ()
vcr = None
class synapse.tests.utils.StreamEvent(*args, **kwargs)[source]

Bases: _io.StringIO, threading.Event

A combination of a io.StringIO object and a threading.Event object.

setMesg(mesg)[source]

Clear the internal event and set a new message that is used to set the event.

Parameters

mesg (str) – The string to monitor for.

Returns

None

write(s)[source]

Write string to file.

Returns the number of characters written, which is always equal to the length of the string.

class synapse.tests.utils.SynTest(*args, **kwargs)[source]

Bases: unittest.case.TestCase

Mark all async test methods as s_glob.synchelp decorated.

Note

This precludes running a single unit test via path using the unittest module.

async addCreatorDeleterRoles(core)[source]

Add two roles to a Cortex proxy, the creator and deleter roles. Creator allows for node:add, prop:set and tag:add actions. Deleter allows for node:del, prop:del and tag:del actions.

Parameters

core – Auth enabled cortex.

addSvcToAha(aha, svcname, ctor, conf=None, dirn=None, provinfo=None)[source]

Creates as service and provision it in a Aha network via the provisioning API.

This assumes the Aha cell has a provision:listen and aha:urls set.

Parameters
  • aha (s_aha.AhaCell) – Aha cell.

  • svcname (str) – Service name.

  • ctor – Service class to add.

  • conf (dict) – Optional service conf.

  • dirn (str) – Optional directory.

  • provinfo (dict)) – Optional provisioning info.

Notes

The config data for the cell is pushed into dirn/cell.yaml. The cells are created with the ctor.anit() function.

async addSvcToCore(svc, core, svcname='svc')[source]

Add a service to a Cortex using telepath over tcp.

async agenlen(x, obj, msg=None)[source]

Assert that the async generator produces x items

async agenraises(exc, gfunc)[source]

Helper to validate that an async generator will throw an exception.

Parameters
  • exc – Exception class to catch

  • gfunc – async Generator

async asyncraises(exc, coro)[source]
checkNode(node, expected)[source]
async checkNodes(core, ndefs)[source]
eq(x, y, msg=None)[source]

Assert X is equal to Y

eqOrNan(x, y, msg=None)[source]

Assert X is equal to Y or they are both NaN (needed since NaN != NaN)

eqish(x, y, places=6, msg=None)[source]

Assert X is equal to Y within places decimal places

async execToolMain(func, argv)[source]
extendOutpFromPatch(outp, patch)[source]

Extend an Outp with lines from a magicMock object from withCliPromptMock.

Parameters
  • outp (TstOutPut) – The outp to extend.

  • patch (mock.MagicMock) – The patch object.

Returns

Returns none.

Return type

None

false(x, msg=None)[source]

Assert X is False

ge(x, y, msg=None)[source]

Assert that X is greater than or equal to Y

genraises(exc, gfunc, *args, **kwargs)[source]

Helper to validate that a generator function will throw an exception.

Parameters
  • exc – Exception class to catch

  • gfunc – Generator function to call.

  • *args – Args passed to the generator function.

  • **kwargs – Kwargs passed to the generator function.

Notes

Wrap a generator function in a list() call and execute that in a bound local using self.raises(exc, boundlocal). The list() will consume the generator until complete or an exception occurs.

getAsyncLoggerStream(logname, mesg='')[source]

Async version of getLoggerStream.

Parameters
  • logname (str) – Name of the logger to get.

  • mesg (str) – A string which, if provided, sets the StreamEvent event if a message containing the string is written to the log.

Notes

The event object mixed in for the AsyncStreamEvent is a asyncio.Event object. This requires the user to await the Event specific calls as neccesary.

Examples

Do an action and wait for a specific log message to be written:

with self.getAsyncLoggerStream('synapse.foo.bar',
                               'big badda boom happened') as stream:
    # Do something that triggers a log message
    await doSomething()
    # Wait for the mesg to be written to the stream
    await stream.wait(timeout=10)

stream.seek(0)
mesgs = stream.read()
# Do something with messages
Returns

An AsyncStreamEvent object.

Return type

AsyncStreamEvent

getHttpSess(auth=None, port=None)[source]

Get an aiohttp ClientSession with a CookieJar.

Parameters
  • auth (str, str) – A tuple of username and password information for http auth.

  • port (int) – Port number to connect to.

Notes

If auth and port are provided, the session will login to a Synapse cell hosted at localhost:port.

Returns

An aiohttp.ClientSession object.

Return type

aiohttp.ClientSession

getLoggerStream(logname, mesg='')[source]

Get a logger and attach a io.StringIO object to the logger to capture log messages.

Parameters
  • logname (str) – Name of the logger to get.

  • mesg (str) – A string which, if provided, sets the StreamEvent event if a message

  • log. (containing the string is written to the) –

Examples

Do an action and get the stream of log messages to check against:

with self.getLoggerStream('synapse.foo.bar') as stream:
    # Do something that triggers a log message
    doSomething()

stream.seek(0)
mesgs = stream.read()
# Do something with messages

Do an action and wait for a specific log message to be written:

with self.getLoggerStream('synapse.foo.bar', 'big badda boom happened') as stream:
    # Do something that triggers a log message
    doSomething()
    stream.wait(timeout=10)  # Wait for the mesg to be written to the stream

stream.seek(0)
mesgs = stream.read()
# Do something with messages

You can also reset the message and wait for another message to occur:

with self.getLoggerStream('synapse.foo.bar', 'big badda boom happened') as stream:
    # Do something that triggers a log message
    doSomething()
    stream.wait(timeout=10)
    stream.setMesg('yo dawg')  # This will now wait for the 'yo dawg' string to be written.
    stream.wait(timeout=10)

stream.seek(0)
mesgs = stream.read()
# Do something with messages

Notes

This only captures logs for the current process.

Yields

StreamEvent – A StreamEvent object

getMagicPromptColors(patch)[source]

Get the colored lines from a MagicMock object from withCliPromptMock.

Parameters

patch (mock.MagicMock) – The MagicMock object from withCliPromptMock.

Returns

A list of tuples, containing color and line data.

Return type

list

getMagicPromptLines(patch)[source]

Get the text lines from a MagicMock object from withCliPromptMock.

Parameters

patch (mock.MagicMock) – The MagicMock object from withCliPromptMock.

Returns

A list of lines.

Return type

list

getRegrAxon(vers, conf=None)[source]
getRegrCore(vers, conf=None)[source]
getRegrDir(*path)[source]
getStructuredAsyncLoggerStream(logname, mesg='')[source]

Async version of getLoggerStream which uses structured logging.

Parameters
  • logname (str) – Name of the logger to get.

  • mesg (str) – A string which, if provided, sets the StreamEvent event if a message containing the string is written to the log.

Notes

The event object mixed in for the AsyncStreamEvent is a asyncio.Event object. This requires the user to await the Event specific calls as needed. The messages written to the stream will be JSON lines.

Examples

Do an action and wait for a specific log message to be written:

with self.getStructuredAsyncLoggerStream('synapse.foo.bar',
                                         '"some JSON string"') as stream:
    # Do something that triggers a log message
    await doSomething()
    # Wait for the mesg to be written to the stream
    await stream.wait(timeout=10)

data = stream.getvalue()
raw_mesgs = [m for m in data.split('\n') if m]
msgs = [json.loads(m) for m in raw_mesgs]
# Do something with messages
Returns

An AsyncStreamEvent object.

Return type

AsyncStreamEvent

getTestAha(conf=None, dirn=None)[source]
getTestAhaProv(conf=None, dirn=None)[source]

Get an Aha cell that is configured for provisioning on aha.loop.vertex.link.

Parameters
  • conf – Optional configuraiton information for the Aha cell.

  • dirn – Optional path to create the Aha cell in.

Returns

The provisioned Aha cell.

Return type

s_aha.AhaCell

getTestAxon(dirn=None, conf=None)[source]

Get a test Axon as an async context manager.

Returns

A Axon object.

Return type

s_axon.Axon

getTestCell(ctor, conf=None, dirn=None)[source]

Get a test Cell.

getTestCertDir(dirn)[source]

Patch the synapse.lib.certdir.certdir singleton and supporting functions with a CertDir instance backed by the provided directory.

Parameters

dirn (str) – The directory used to back the new CertDir singleton.

Returns

The patched CertDir object that is the current singleton.

Return type

s_certdir.CertDir

getTestConfDir(name, conf=None)[source]
getTestCore(conf=None, dirn=None)[source]

Get a simple test Cortex as an async context manager.

Returns

A Cortex object.

Return type

s_cortex.Cortex

getTestCoreAndProxy(conf=None, dirn=None)[source]

Get a test Cortex and the Telepath Proxy to it.

Returns

The Cortex and a Proxy representing a CoreApi object.

Return type

(s_cortex.Cortex, s_cortex.CoreApi)

getTestCoreProxSvc(ssvc, ssvc_conf=None, core_conf=None)[source]

Get a test Cortex, the Telepath Proxy to it, and a test service instance.

Parameters
  • ssvc – Ctor to the Test Service.

  • ssvc_conf – Service configuration.

  • core_conf – Cortex configuration.

Returns

The Cortex, Proxy, and service instance.

Return type

(s_cortex.Cortex, s_cortex.CoreApi, testsvc)

getTestCryo(dirn=None, conf=None)[source]

Get a simple test Cryocell as an async context manager.

Returns

Test cryocell.

Return type

s_cryotank.CryoCell

getTestCryoAndProxy(dirn=None)[source]

Get a test Cryocell and the Telepath Proxy to it.

Returns

CryoCell, s_cryotank.CryoApi): The CryoCell and a Proxy representing a CryoApi object.

Return type

(s_cryotank

getTestDir(mirror=None, copyfrom=None, chdir=False, startdir=None)[source]

Get a temporary directory for test purposes. This destroys the directory afterwards.

Parameters
  • mirror (str) – A directory to mirror into the test directory.

  • startdir (str) – The directory under which to place the temporary kdirectory

Notes

The mirror argument is normally used to mirror test directory under synapse/tests/files. This is accomplished by passing in the name of the directory (such as testcore) as the mirror argument.

If the mirror argument is an absolute directory, that directory will be copied to the test directory.

Returns

The path to a temporary directory.

Return type

str

getTestDmon()[source]
getTestFilePath(*names)[source]
getTestHive()[source]
getTestHiveDmon()[source]
getTestHiveFromDirn(dirn)[source]
getTestJsonStor(dirn=None, conf=None)[source]
getTestOutp()[source]

Get a Output instance with a expects() function.

Returns

A TstOutPut instance.

Return type

TstOutPut

getTestProxy(dmon, name, **kwargs)[source]
getTestReadWriteCores(conf=None, dirn=None)[source]

Get a read/write core pair.

Notes

By default, this returns the same cortex. It is expected that a test which needs two distinct Cortexes implements the bridge themselves.

Returns

A tuple of Cortex objects.

Return type

(s_cortex.Cortex, s_cortex.Cortex)

getTestSynDir()[source]

Combines getTestDir() and setSynDir() into one.

getTestTeleHive()[source]
getTestUrl(dmon, name, **opts)[source]
gt(x, y, msg=None)[source]

Assert that X is greater than Y

isin(member, container, msg=None)[source]

Assert a member is inside of a container.

isinstance(obj, cls, msg=None)[source]

Assert a object is the instance of a given class or tuple of classes.

istufo(obj)[source]

Check to see if an object is a tufo.

Parameters

obj (object) – Object being inspected.

Notes

This does not make any assumptions about the contents of the dictionary. This validates the object to be a tuple of length two, containing a str or None as the first value, and a dict as the second value.

Returns

None

le(x, y, msg=None)[source]

Assert that X is less than or equal to Y

len(x, obj, msg=None)[source]

Assert that the length of an object is equal to X

lt(x, y, msg=None)[source]

Assert that X is less than Y

ne(x, y)[source]

Assert X is not equal to Y

nn(x, msg=None)[source]

Assert X is not None

none(x, msg=None)[source]

Assert X is None

noprop(info, prop)[source]

Assert a property is not present in a dictionary.

notin(member, container, msg=None)[source]

Assert a member is not inside of a container.

printed(msgs, text)[source]
raises(*args, **kwargs)[source]

Assert a function raises an exception.

redirectStdin(new_stdin)[source]

Temporary replace stdin.

Parameters

new_stdin (file-like object) – file-like object.

Examples

Patch stdin with a string buffer:

inp = io.StringIO('stdin stuff\nanother line\n')
with self.redirectStdin(inp):
    main()

Here’s a way to use this for code that’s expecting the stdin buffer to have bytes:

inp = Mock()
inp.buffer = io.BytesIO(b'input data')
with self.redirectStdin(inp):
    main()
Returns

None

async runCoreNodes(core, query, opts=None)[source]

Run a storm query through a Cortex as a SchedCoro and return the results.

setSynDir(dirn)[source]

Sets s_common.syndir to a specific directory and then unsets it afterwards.

Parameters

dirn (str) – Directory to set syndir to.

Notes

This is to be used as a context manager.

setTstEnvars(**props)[source]

Set Environment variables for the purposes of running a specific test.

Parameters
  • **props – A kwarg list of envars to set. The values set are run

  • strings. (through str() to ensure we're setting) –

Examples

Run a test while a envar is set:

with self.setEnvars(magic='haha') as nop:
    ret = dostuff()
    self.true(ret)

Notes

This helper explicitly sets and unsets values in os.environ, as os.putenv does not automatically updates the os.environ object.

Yields

None. This context manager yields None. Upon exiting, envars are either removed from os.environ or reset to their previous values.

skip(mesg)[source]
skipIfNexusReplay()[source]

Allow skipping a test if SYNDEV_NEXUS_REPLAY envar is set.

Raises

unittest.SkipTest if SYNDEV_NEXUS_REPLAY envar is set to true value.

skipIfNoInternet()[source]

Allow skipping a test if SYN_TEST_SKIP_INTERNET envar is set.

Raises

unittest.SkipTest if SYN_TEST_SKIP_INTERNET envar is set to a integer greater than 1.

skipLongTest()[source]

Allow skipping a test if SYN_TEST_SKIP_LONG envar is set.

Raises

unittest.SkipTest if SYN_TEST_SKIP_LONG envar is set to a integer greater than 1.

sorteq(x, y, msg=None)[source]

Assert two sorted sequences are the same.

stablebuid(valu=None)[source]

A stable buid generation for testing purposes

stableguid(valu=None)[source]

A stable guid generation for testing purposes

stormHasNoErr(mesgs)[source]

Raise an AssertionError if there is a message of type “err” in the list.

Parameters

mesgs (list) – A list of storm messages.

stormHasNoWarnErr(mesgs)[source]

Raise an AssertionError if there is a message of type “err” or “warn” in the list.

Parameters

mesgs (list) – A list of storm messages.

stormIsInErr(mesg, mesgs)[source]

Check if a string is present in all of the error messages from a stream of storm messages.

Parameters
  • mesg (str) – A string to check.

  • mesgs (list) – A list of storm messages.

stormIsInPrint(mesg, mesgs)[source]

Check if a string is present in all of the print messages from a stream of storm messages.

Parameters
  • mesg (str) – A string to check.

  • mesgs (list) – A list of storm messages.

stormIsInWarn(mesg, mesgs)[source]

Check if a string is present in all of the warn messages from a stream of storm messages.

Parameters
  • mesg (str) – A string to check.

  • mesgs (list) – A list of storm messages.

stormNotInPrint(mesg, mesgs)[source]

Assert a string is not present in all of the print messages from a stream of storm messages.

Parameters
  • mesg (str) – A string to check.

  • mesgs (list) – A list of storm messages.

thisHostMust(**props)[source]

Requires a host having a specific property.

Parameters

**props

Raises

unittest.SkipTest if the required property is missing.

thisHostMustNot(**props)[source]

Requires a host to not have a specific property.

Parameters

**props

Raises

unittest.SkipTest if the required property is missing.

true(x, msg=None)[source]

Assert X is True

withCliPromptMock()[source]

Context manager to mock our use of Prompt Toolkit’s print_formatted_text function.

Returns

Yields a mock.MagikMock object.

Return type

mock.MagicMock

withCliPromptMockExtendOutp(outp)[source]

Context manager to mock our use of Prompt Toolkit’s print_formatted_text function and extend the lines to an an output object.

Parameters

outp (TstOutPut) – The outp to extend.

Notes

This extends the outp with the lines AFTER the context manager has exited.

Returns

Yields a mock.MagicMock object.

Return type

mock.MagicMock

withNexusReplay(replay=False)[source]

Patch so that the Nexus apply log is applied twice. Useful to verify idempotency.

Parameters

replay (bool) – Set the default value of resolving the existence of SYNDEV_NEXUS_REPLAY variable. This can be used to force the apply patch without using the environment variable.

Notes

This is applied if the environment variable SYNDEV_NEXUS_REPLAY is set to a non zero value or the replay argument is set to True.

Returns

An exitstack object.

Return type

contextlib.ExitStack

withSetLoggingMock()[source]

Context manager to mock calls to the setlogging function to avoid unittests calling logging.basicconfig.

Returns

Yields a mock.MagikMock object.

Return type

mock.MagicMock

withStableUids()[source]

A context manager that generates guids and buids in sequence so that successive test runs use the same data

withTestCmdr(cmdg)[source]
class synapse.tests.utils.TestCmd(runt, runtsafe)[source]

Bases: synapse.lib.storm.Cmd

A test command

async execStormCmd(runt, genr)[source]

Abstract base method

forms = {'input': ['test:str', 'inet:ipv6'], 'nodedata': [('foo', 'inet:ipv4'), ('bar', 'inet:fqdn')], 'output': ['inet:fqdn']}
getArgParser()[source]
name = 'testcmd'
class synapse.tests.utils.TestModule(core, conf=None)[source]

Bases: synapse.lib.module.CoreModule

async addTestRecords(snap, items)[source]
getModelDefs()[source]
getStormCmds()[source]

Module implementers may override this to provide a list of Storm commands which will be loaded into the Cortex.

Returns

A list of Storm Command classes (not instances).

Return type

list

async initCoreModule()[source]

Module implementers may override this method to initialize the module after the Cortex has completed and is accessible to perform storage operations.

Notes

This is the preferred function to override for implementing custom code that needs to be executed during Cortex startup.

Any exception raised within this method will remove the module from the list of currently loaded modules.

This is called for modules after getModelDefs() and getStormCmds() has been called, in order to allow for model loading and storm command loading prior to code execution offered by initCoreModule.

A failure during initCoreModule will not unload data model or storm commands registered by the module.

Returns

None

testguid = '8f1401de15918358d5247e21ca29a814'
class synapse.tests.utils.TestRunt(name, **kwargs)[source]

Bases: object

getStorNode(form)[source]
class synapse.tests.utils.TestSubType(modl, name, info, opts)[source]

Bases: synapse.lib.types.Type

norm(valu)[source]

Normalize the value for a given type.

Parameters

valu (obj) – The value to normalize.

Returns

The normalized valu, info tuple.

Return type

((obj,dict))

Notes

The info dictionary uses the following key conventions:

subs (dict): The normalized sub-fields as name: valu entries.

repr(norm)[source]

Return a printable representation for the value. This may return a string or a tuple of values for display purposes.

stortype: int = 4
class synapse.tests.utils.TestType(modl, name, info, opts)[source]

Bases: synapse.lib.types.Type

postTypeInit()[source]
stortype: int = 1
class synapse.tests.utils.ThreeType(modl, name, info, opts)[source]

Bases: synapse.lib.types.Type

norm(valu)[source]

Normalize the value for a given type.

Parameters

valu (obj) – The value to normalize.

Returns

The normalized valu, info tuple.

Return type

((obj,dict))

Notes

The info dictionary uses the following key conventions:

subs (dict): The normalized sub-fields as name: valu entries.

repr(valu)[source]

Return a printable representation for the value. This may return a string or a tuple of values for display purposes.

stortype: int = 2
class synapse.tests.utils.TstEnv[source]

Bases: object

add(name, item, fini=False)[source]
async fini()[source]
class synapse.tests.utils.TstOutPut[source]

Bases: synapse.lib.output.OutPutStr

clear()[source]
expect(substr, throw=True)[source]

Check if a string is present in the messages captured by the OutPutStr object.

Parameters
  • substr (str) – String to check for the existence of.

  • throw (bool) – If True, a missing substr results in a Exception being thrown.

Returns

True if the string is present; False if the string is not present and throw is False.

Return type

bool

async synapse.tests.utils.alist(coro)[source]
synapse.tests.utils.norm(z)[source]