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)

Bases: StringIO, Event

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

jsonlines() → List[dict]

Get the messages as jsonlines. May throw Json errors if the captured stream is not jsonlines.

setMesg(mesg)

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)

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)

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)

Bases: object

addCmd(cmd)

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)

Bases: CoreModule

getModelDefs()

class synapse.tests.utils.HttpReflector(application: Application, request: HTTPServerRequest, **kwargs: Any)

Bases: Handler

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

async get()

async head()

async post()

class synapse.tests.utils.LibDepr(runt, name=())

Bases: Lib

Deprecate me!

addLibFuncs()

async boop(valu)

class synapse.tests.utils.LibTst(runt, name=())

Bases: Lib

LibTst for testing!

addLibFuncs()

async beep(valu)

Example storm func

async someargs(valu, bar=True, faz=None)

Example storm func

class synapse.tests.utils.PickleableMagicMock(*args, **kw)

Bases: MagicMock

class synapse.tests.utils.ReloadCell

Bases: Cell

async addTestBadReload()

async addTestReload()

async getCellInfo()

Return metadata specific for the Cell.

Notes

By default, this function returns information about the base Cell implementation, which reflects the base information in the Synapse Cell.

It is expected that implementers override the following Class attributes in order to provide meaningful version information:

COMMIT - A Git Commit VERSION - A Version tuple. VERSTRING - A Version string.

Returns:

A Dictionary of metadata.

Return type:

Dict

async postAnit()

Method called after self.__anit__() has completed, but before anit() returns the object to the caller.

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

Bases: SynTest

assetdir = None

getTestCore(conf=None, dirn=None, prepkghook=None)

Get a simple test Cortex as an async context manager.

Returns:

A Cortex object.

Return type:

s_cortex.Cortex

async initTestCore(core)

This is executed after the package has been loaded and a VCR context has been entered.

pkgprotos = ()

vcr = None

class synapse.tests.utils.StreamEvent(*args, **kwargs)

Bases: StringIO, Event

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

jsonlines() → List[dict]

Get the messages as jsonlines. May throw Json errors if the captured stream is not jsonlines.

setMesg(mesg)

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)

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)

Bases: 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)

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)

Creates a service and provisions it in an 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')

Add a service to a Cortex using telepath over tcp.

async agenlen(x, obj, msg=None)

Assert that the async generator produces x items

async agenraises(exc, gfunc)

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

Parameters:

• exc – Exception class to catch

• gfunc – async Generator

async asyncraises(exc, coro)

checkNode(node, expected)

async checkNodes(core, ndefs)

eq(x, y, msg=None)

Assert X is equal to Y

eqOrNan(x, y, msg=None)

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

eqish(x, y, places=6, msg=None)

Assert X is equal to Y within places decimal places

async execToolMain(func, argv)

extendOutpFromPatch(outp, patch)

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)

Assert X is False

ge(x, y, msg=None)

Assert that X is greater than or equal to Y

genraises(exc, gfunc, *args, **kwargs)

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='') → AbstractContextManager[StreamEvent, None, None]

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

getCellConf(conf=None)

getHttpSess(auth=None, port=None)

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='')

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)

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)

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)

getRegrCore(vers, conf=None, maxvers=None)

getRegrDir(*path)

getStructuredAsyncLoggerStream(logname, mesg='') → AbstractContextManager[AsyncStreamEvent, None, None]

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)

msgs = stream.jsonlines()
# Do something with messages

Returns:

An AsyncStreamEvent object.

Return type:

AsyncStreamEvent

getTestAha(conf=None, dirn=None, ctor=None)

getTestAxon(dirn=None, conf=None)

Get a test Axon as an async context manager.

Returns:

A Axon object.

Return type:

s_axon.Axon

getTestCell(ctor=<class 'synapse.lib.cell.Cell'>, conf=None, dirn=None)

Get a test Cell.

getTestCertDir(dirn)

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)

getTestCore(conf=None, dirn=None)

Get a simple test Cortex as an async context manager.

Returns:

A Cortex object.

Return type:

s_cortex.Cortex

getTestCoreAndProxy(conf=None, dirn=None)

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)

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)

Get a simple test Cryocell as an async context manager.

Returns:

Test cryocell.

Return type:

s_cryotank.CryoCell

getTestCryoAndProxy(dirn=None)

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) → AbstractContextManager[str, None, None]

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

Parameters:

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

• copyfrom (str) – An arbitrary directory to copy into the test directory.

• chdir (boolean) – If true, chdir the current process to that directory. This is undone when the context manager exits.

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

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()

getTestFilePath(*names)

getTestHive()

getTestHiveDmon()

getTestHiveFromDirn(dirn)

getTestJsonStor(dirn=None, conf=None)

getTestOutp()

Get a Output instance with a expects() function.

Returns:

A TstOutPut instance.

Return type:

TstOutPut

getTestProxy(dmon, name, **kwargs)

getTestReadWriteCores(conf=None, dirn=None)

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()

Combines getTestDir() and setSynDir() into one.

getTestTeleHive()

getTestUrl(dmon, name, **opts)

gt(x, y, msg=None)

Assert that X is greater than Y

isin(member, container, msg=None)

Assert a member is inside of a container.

isinstance(obj, cls, msg=None)

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

istufo(obj)

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)

Assert that X is less than or equal to Y

len(x, obj, msg=None)

Assert that the length of an object is equal to X

lt(x, y, msg=None)

Assert that X is less than Y

mayTestDir(dirn: str | None) → AbstractContextManager[str, None, None]

Convenience method to make a temporary directory or use an existing one.

Parameters:

dirn – Directory to use, or None.

Returns:

The directory to use.

ne(x, y)

Assert X is not equal to Y

nn(x, msg=None)

Assert X is not None

none(x, msg=None)

Assert X is None

noprop(info, prop)

Assert a property is not present in a dictionary.

notin(member, container, msg=None)

Assert a member is not inside of a container.

printed(msgs, text)

raises(*args, **kwargs)

Assert a function raises an exception.

redirectStdin(new_stdin)

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)

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

setSynDir(dirn)

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)

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)

skipIfNexusReplay()

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()

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. –

skipIfNoPath(path, mesg=None)

Allows skipping a test if the test/files path does not exist.

Parameters:

• path (str) – Path to check.

• mesg (str) – Optional additional message.

Raises:

unittest.SkipTest if the path does not exist. –

skipLongTest()

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)

Assert two sorted sequences are the same.

stablebuid(valu=None)

A stable buid generation for testing purposes

stableguid(valu=None)

A stable guid generation for testing purposes

stormHasNoErr(mesgs)

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

Parameters:

mesgs (list) – A list of storm messages.

stormHasNoWarnErr(mesgs)

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)

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, deguid=False, whitespace=True)

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)

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)

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.

stormNotInWarn(mesg, mesgs)

Assert a string is not 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.

thisEnvMust(*envvars)

Requires a host must have environment variables set to truthy values.

Parameters:

*envars – Environment variables to require being present.

thisEnvMustNot(*envvars)

Requires a host must not have environment variables set to truthy values.

Parameters:

*envars – Environment variables to require being absent or set to falsey values.

thisHostMust(**props)

Requires a host having a specific property.

Parameters:

**props

Raises:

unittest.SkipTest if the required property is missing. –

thisHostMustNot(**props)

Requires a host to not have a specific property.

Parameters:

**props

Raises:

unittest.SkipTest if the required property is missing. –

true(x, msg=None)

Assert X is True

withCliPromptMock()

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)

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)

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()

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

Returns:

Yields a mock.MagicMock object.

Return type:

mock.MagicMock

withStableUids()

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

withTestCmdr(cmdg)

class synapse.tests.utils.TestCmd(runt, runtsafe)

Bases: Cmd

A test command

async execStormCmd(runt, genr)

Abstract base method

forms = {'input': ['test:str', 'inet:ipv6'], 'nodedata': [('foo', 'inet:ipv4'), ('bar', 'inet:fqdn')], 'output': ['inet:fqdn']}

getArgParser()

name = 'testcmd'

class synapse.tests.utils.TestModule(core, conf=None)

Bases: CoreModule

async addTestRecords(snap, items)

getModelDefs()

getStormCmds()

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()

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)

Bases: object

getStorNode(form)

class synapse.tests.utils.TestSubType(modl, name, info, opts)

Bases: Type

norm(valu)

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)

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)

Bases: Type

postTypeInit()

stortype: int = 1

class synapse.tests.utils.ThreeType(modl, name, info, opts)

Bases: Type

norm(valu)

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)

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

Bases: object

add(name, item, fini=False)

async fini()

class synapse.tests.utils.TstOutPut

Bases: OutPutStr

clear()

expect(substr, throw=True, whitespace=True)

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)

synapse.tests.utils.deguidify(x)

synapse.tests.utils.jsonlines(text: str)

synapse.tests.utils.matchContexts(testself)

synapse.tests.utils.norm(z)

async synapse.tests.utils.waitForBehold(core, events)