You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
mitogen/tests/push_file_service_test.py

57 lines
1.4 KiB
Python

import os
import tempfile
import unittest2
import mitogen.core
import mitogen.service
import testlib
from mitogen.core import b
def prepare():
# ensure module loading delay is complete before loading PushFileService.
pass
@mitogen.core.takes_router
def wait_for_file(path, router):
pool = mitogen.service.get_or_create_pool(router=router)
service = pool.get_service(u'mitogen.service.PushFileService')
return service.get(path)
class PropagateToTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.service.PushFileService
def test_two_grandchild_one_intermediary(self):
tf = tempfile.NamedTemporaryFile()
path = mitogen.core.to_text(tf.name)
try:
tf.write(b('test'))
tf.flush()
interm = self.router.local(name='interm')
c1 = self.router.local(via=interm, name='c1')
c2 = self.router.local(via=interm)
c1.call(prepare)
c2.call(prepare)
service = self.klass(router=self.router)
service.propagate_to(context=c1, path=path)
service.propagate_to(context=c2, path=path)
s = c1.call(wait_for_file, path=path)
self.assertEquals(b('test'), s)
s = c2.call(wait_for_file, path=path)
self.assertEquals(b('test'), s)
finally:
tf.close()
if __name__ == '__main__':
unittest2.main()