本文整理汇总了Python中tempdir.TempDir类的典型用法代码示例。如果您正苦于以下问题：Python TempDir类的具体用法？Python TempDir怎么用？Python TempDir使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了TempDir类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_use_build_script_instead_of_docker_file_if_available
def test_use_build_script_instead_of_docker_file_if_available(self, docker_mock, res_mock, tempDir_mock):
# given
provider = DockerProvider(self._adapter, 'leap_provider', self._leap_provider_x509)
tempBuildDir = TempDir()
try:
tempDir_mock.return_value = tempBuildDir
tempBuildDir_name = tempBuildDir.name
with NamedTemporaryFile() as file:
res_mock.resource_exists.return_value = True
res_mock.resource_string.return_value = '#!/bin/bash\necho %s $PWD > %s' % (file.name, file.name)
# when
provider.initialize()
# then
res_mock.resource_exists.assert_called_with('pixelated.resources', 'init-pixelated-docker-context.sh')
res_mock.resource_string.assert_called_with('pixelated.resources', 'init-pixelated-docker-context.sh')
with open(file.name, "r") as input:
data = input.read().replace('\n', '')
self.assertEqual('%s %s' % (file.name, os.path.realpath(tempBuildDir_name)), data)
docker_mock.return_value.build.assert_called_once_with(path=tempBuildDir_name, tag='pixelated:latest', fileobj=None)
finally:
tempBuildDir.dissolve()
开发者ID:Meistache，项目名称:pixelated-dispatcher，代码行数:25，代码来源:docker_provider_test.py
示例2: SearchEngineTest
class SearchEngineTest(unittest.TestCase):
def setUp(self):
self.tempdir = TempDir()
self.agent_home = self.tempdir.name
def tearDown(self):
self.tempdir.dissolve()
def test_encoding(self):
# given
se = SearchEngine(INDEX_KEY, self.agent_home)
headers = {
"From": "[email protected]",
"To": "=?utf-8?b?IsOEw7zDtiDDlsO8w6QiIDxmb2xrZXJAcGl4ZWxhdGVkLXByb2plY3Qub3Jn?=\n =?utf-8?b?PiwgRsO2bGtlciA8Zm9sa2VyQHBpeGVsYXRlZC1wcm9qZWN0Lm9yZz4=?=",
"Cc": "=?utf-8?b?IsOEw7zDtiDDlsO8w6QiIDxmb2xrZXJAcGl4ZWxhdGVkLXByb2plY3Qub3Jn?=\n =?utf-8?b?PiwgRsO2bGtlciA8Zm9sa2VyQHBpeGVsYXRlZC1wcm9qZWN0Lm9yZz4=?=",
"Subject": "Some test mail",
}
# when
se.index_mail(test_helper.pixelated_mail(extra_headers=headers, chash="mailid"))
result = se.search("folker")
self.assertEqual((["mailid"], 1), result)
开发者ID:EderRoger，项目名称:pixelated-user-agent，代码行数:25，代码来源:test_search.py
示例3: SearchEngineTest
class SearchEngineTest(unittest.TestCase):
def setUp(self):
self.tempdir = TempDir()
self.agent_home = self.tempdir.name
def tearDown(self):
self.tempdir.dissolve()
def test_encoding(self):
# given
se = SearchEngine(INDEX_KEY, self.agent_home)
headers = {
'From': '[email protected]',
'To': '=?utf-8?b?IsOEw7zDtiDDlsO8w6QiIDxmb2xrZXJAcGl4ZWxhdGVkLXByb2plY3Qub3Jn?=\n =?utf-8?b?PiwgRsO2bGtlciA8Zm9sa2VyQHBpeGVsYXRlZC1wcm9qZWN0Lm9yZz4=?=',
'Cc': '=?utf-8?b?IsOEw7zDtiDDlsO8w6QiIDxmb2xrZXJAcGl4ZWxhdGVkLXByb2plY3Qub3Jn?=\n =?utf-8?b?PiwgRsO2bGtlciA8Zm9sa2VyQHBpeGVsYXRlZC1wcm9qZWN0Lm9yZz4=?=',
'Subject': 'Some test mail',
}
# when
se.index_mail(LeapMail('mailid', 'INBOX', headers=headers)) # test_helper.pixelated_mail(extra_headers=headers, chash='mailid'))
result = se.search('folker')
self.assertEqual((['mailid'], 1), result)
开发者ID:rdoh，项目名称:pixelated-user-agent，代码行数:25，代码来源:test_search.py
示例4: SearchEngineTest
class SearchEngineTest(unittest.TestCase):
def setUp(self):
self.tempdir = TempDir()
self.agent_home = self.tempdir.name
def tearDown(self):
self.tempdir.dissolve()
def test_index_mail_secured_by_lock(self):
# given
soledad_querier = mock()
lock_stub = LockStub()
when(soledad_querier).get_index_masterkey().thenReturn(INDEX_KEY)
self.assertEqual(INDEX_KEY, soledad_querier.get_index_masterkey())
se = SearchEngine(soledad_querier, self.agent_home)
se._write_lock = lock_stub
headers = {
'From': '[email protected]',
'To': '[email protected]',
'Subject': 'Some test mail',
}
# when
se.index_mail(test_helper.pixelated_mail(extra_headers=headers))
# then
self.assertTrue(lock_stub.called)
def test_encoding(self):
# given
soledad_querier = mock()
when(soledad_querier).get_index_masterkey().thenReturn(INDEX_KEY)
se = SearchEngine(soledad_querier, self.agent_home)
headers = {
'From': '[email protected]',
'To': '=?utf-8?b?IsOEw7zDtiDDlsO8w6QiIDxmb2xrZXJAcGl4ZWxhdGVkLXByb2plY3Qub3Jn?=\n =?utf-8?b?PiwgRsO2bGtlciA8Zm9sa2VyQHBpeGVsYXRlZC1wcm9qZWN0Lm9yZz4=?=',
'Cc': '=?utf-8?b?IsOEw7zDtiDDlsO8w6QiIDxmb2xrZXJAcGl4ZWxhdGVkLXByb2plY3Qub3Jn?=\n =?utf-8?b?PiwgRsO2bGtlciA8Zm9sa2VyQHBpeGVsYXRlZC1wcm9qZWN0Lm9yZz4=?=',
'Subject': 'Some test mail',
}
# when
se.index_mail(test_helper.pixelated_mail(extra_headers=headers, chash='mailid'))
result = se.search('folker')
self.assertEqual((['mailid'], 1), result)
开发者ID:amukiza，项目名称:pixelated-user-agent，代码行数:50，代码来源:test_search.py
示例5: setUp
def setUp(self):
self.users = MagicMock(spec=Users)
self._tmpdir = TempDir()
self.root_path = self._tmpdir.name
self._adapter = MagicMock(wraps=PixelatedDockerAdapter())
self._adapter.docker_image_name.return_value = 'pixelated'
self._leap_provider_x509 = LeapProviderX509Info()
开发者ID:xenobyte，项目名称:pixelated-dispatcher，代码行数:7，代码来源:docker_provider_test.py
示例6: setUp
def setUp(self):
self.d = TempDir()
self.keyfile_path = os.path.join(self.d.name, 'keyfile')
self.database_path = os.path.join(self.d.name, 'database')
self.keyfile = K.create(
self.keyfile_path,
self.database_path)
D.create(self.keyfile)
开发者ID:WilliamMayor，项目名称:pinscher，代码行数:8，代码来源:test_database.py
示例7: setUp
def setUp(self):
self.mock_provider = RESTfulServerTest.mock_provider
self.mock_provider.reset_mock()
self.mock_users.reset_mock()
self.mock_authenticator.reset_mock()
self.ssl_request = requests.Session()
self.ssl_request.mount('https://', EnforceTLSv1Adapter())
self._tmpdir = TempDir()
self._root_path = self._tmpdir.name
开发者ID:Meistache，项目名称:pixelated-dispatcher，代码行数:11，代码来源:manager_test.py
示例8: __enter__
def __enter__(self):
self.tempdir_obj = TempDir()
tempdir = self.tempdir_obj.__enter__()
logfile = join(tempdir, self.description + "out")
os.mkfifo(logfile)
bash_command = "{} 2>&1 | tee -i --output-error=warn {} | tee -i --output-error=warn {}_log.txt".format(
self.spawn_cmd, pipes.quote(logfile), self.description)
self.pane = check_output([
'tmux', 'split-window', '-hdP',
'-F', '#{pane_id}', '-t', pane,
'bash', '-c', bash_command]).strip()
self.proc = Expect(os.open(logfile, os.O_RDONLY | os.O_NONBLOCK), quiet=True)
self.proc.expect("(rr) ")
return self
开发者ID:mewbak，项目名称:panda，代码行数:16，代码来源:diverge.py
示例9: ForkProviderTest
class ForkProviderTest(unittest.TestCase):
def setUp(self):
self._tmpdir = TempDir()
self.root_path = self._tmpdir.name
self.runner = MagicMock(spec=Adapter)
self.provider = ForkProvider(self.root_path, self.runner)
self.provider.initialize()
def tearDown(self):
self._tmpdir.dissolve()
@patch('random.getrandbits')
def test_new_instance_can_be_added(self, randmock):
hashed_password = '1298e6969236f8efe078c2718acc793c6459b6d94720eb2443a9d04820ac68faacc0d79ba28c7b323be13a69d30bf19340cfcadab79c9f11f8b7388436409de9'
hex_salt = 31
randmock.return_value = 1
self.provider.add('test', 'password')
instance_path = join(self.root_path, 'test')
cfg_path = join(instance_path, 'agent.cfg')
data_path = join(instance_path, 'data')
gnupg_path = join(instance_path, 'gnupg')
self.assertTrue(isfile(cfg_path), 'No config file created')
self.assertTrue(isdir(instance_path), 'No folder for user has been created')
self.assertTrue(isdir(data_path), 'No maipile folder for user has been created')
self.assertTrue(isdir(gnupg_path), 'No gnupg folder for user has been created')
self.assert_config_file(cfg_path, 'test', hashed_password, hex_salt)
self.assertEqual(stat.S_IFDIR | stat.S_IRWXU, os.stat(instance_path).st_mode)
self.assertEqual(stat.S_IFDIR | stat.S_IRWXU, os.stat(data_path).st_mode)
self.assertEqual(stat.S_IFDIR | stat.S_IRWXU, os.stat(gnupg_path).st_mode)
@patch('random.getrandbits')
def test_random_salt_is_used(self, randmock):
hashed_password = 'ecb5c3382726fd831a54bc888d46c8813a18750220c2067cd7bc3fe4f7791261099426cabc6ad07220080943de2c331afddd75210d3ecdd7c5742c85146ace3c'
salt = 2
hex_salt = 32
randmock.return_value = salt
instance_path = join(self.root_path, 'test')
cfg_path = join(instance_path, 'agent.cfg')
self.provider.add('test', 'password')
self.assert_config_file(cfg_path, 'test', hashed_password, hex_salt)
def test_multiple_instances_can_be_added(self):
self.provider.add('first', 'password')
self.provider.add('second', 'password')
self.assertEqual(['first', 'second'], self.provider.list())
def test_instances_can_not_be_added_twice(self):
self.provider.add('test', 'password')
self.assertRaises(InstanceAlreadyExistsError, self.provider.add, 'test', 'password')
def test_remove_raises_exception_if_instance_does_not_exist(self):
self.assertRaises(ValueError, self.provider.remove, 'test')
def test_instances_can_be_removed(self):
self.provider.add('test', 'password')
self.provider.remove('test')
def test_that_non_existing_instance_cannot_be_started(self):
self.assertRaises(InstanceNotFoundError, self.provider.start, 'test')
def test_that_instance_can_be_started_and_gets_initialized(self):
self.provider.add('test', 'password')
self.provider.start('test')
self.runner.initialize.assert_called_with('test')
self.runner.start.assert_called_with('test')
def test_that_instance_cannot_be_started_twice(self):
self.provider.add('test', 'password')
self.provider.start('test')
self.assertRaises(InstanceAlreadyRunningError, self.provider.start, 'test')
def test_that_running_instances_are_in_runnig_list(self):
self._init_runner_memory_usage()
for name in ['one', 'two', 'three']:
self.provider.add(name, 'password')
self.provider.start(name)
self.provider.add('not-started', 'password')
self.assertEqual({'one', 'two', 'three'}, set(self.provider.list_running()))
def test_that_non_existing_instance_cannot_be_stopped(self):
self.assertRaises(InstanceNotRunningError, self.provider.stop, 'test')
def test_that_non_started_instance_cannot_be_stopped(self):
self.provider.add('test', 'password')
self.assertRaises(InstanceNotRunningError, self.provider.stop, 'test')
def test_that_running_instance_can_be_stopped(self):
#.........这里部分代码省略.........
开发者ID:mmircea16，项目名称:pixelated-dispatcher，代码行数:101，代码来源:provider_test.py
示例10: RRInstance
class RRInstance(object):
def __init__(self, description, rr_replay, source_pane):
self.description = description
self.spawn_cmd = "{} replay {}".format(
pipes.quote(cli_args.rr), pipes.quote(rr_replay))
self.source_pane = source_pane
self.breakpoints = {}
self.watches_set = 0
def __repr__(self):
return "RRInstance({!r})".format(self.description)
# Runs in child process.
def sendline(self, msg):
check_call(['tmux', 'send-keys', '-t', self.pane, '-l', msg])
check_call(['tmux', 'send-keys', '-t', self.pane, 'ENTER'])
# Runs in child process.
def kill(self):
check_call(['tmux', 'kill-pane', '-t', self.pane])
def __enter__(self):
self.tempdir_obj = TempDir()
tempdir = self.tempdir_obj.__enter__()
logfile = join(tempdir, self.description + "out")
os.mkfifo(logfile)
bash_command = "{} 2>&1 | tee -i --output-error=warn {} | tee -i --output-error=warn {}_log.txt".format(
self.spawn_cmd, pipes.quote(logfile), self.description)
self.pane = check_output([
'tmux', 'split-window', '-hdP',
'-F', '#{pane_id}', '-t', pane,
'bash', '-c', bash_command]).strip()
self.proc = Expect(os.open(logfile, os.O_RDONLY | os.O_NONBLOCK), quiet=True)
self.proc.expect("(rr) ")
return self
def __exit__(self, exc_type, exc_value, traceback):
if exc_type:
self.kill()
self.tempdir_obj.__exit__(exc_type, exc_value, traceback)
def gdb(self, *args, **kwargs):
timeout = kwargs.get('timeout', None)
cmd = " ".join(map(str, args))
print "(rr-{}) {}".format(self.description, cmd)
sys.stdout.flush()
while True:
try:
os.read(self.proc.fd, 1024)
except OSError as e:
if e.errno in [EAGAIN, EWOULDBLOCK]:
break
else:
raise
self.sendline(cmd)
try:
output = self.proc.expect("(rr) ", timeout=timeout)
except TimeoutExpired:
print self.proc.sofar
print "EXCEPTION!"
sys.stdout.flush()
return output
def quit(self):
self.gdb("set confirm off")
self.sendline("quit")
def breakpoint(self, break_arg):
result = self.gdb("break", break_arg)
bp_num = int(re.search(r"Breakpoint ([0-9]+) at", result).group(1))
self.breakpoints[break_arg] = bp_num
def disable_all(self):
self.gdb("disable")
def enable(self, break_arg):
self.gdb("enable", self.breakpoints[break_arg])
def condition(self, break_arg, cond):
self.gdb("condition", self.breakpoints[break_arg], cond)
def condition_instr(self, break_arg, op, instr):
if not hasattr(self, 'instr_count_ptr'):
self.instr_count_ptr = self.get_value(
"&cpus->tqh_first->rr_guest_instr_count")
self.condition(
break_arg, "*(uint64_t *){} {} {}".format(self.instr_count_ptr, op, instr))
def get_value(self, value_str):
result = self.gdb("print/u", value_str)
re_result = re.search(r"\$[0-9]+ = ([0-9]+)", result)
if re_result:
return long(re_result.group(1))
else:
#.........这里部分代码省略.........
开发者ID:mewbak，项目名称:panda，代码行数:101，代码来源:diverge.py
示例11: setUp
def setUp(self):
self._tmpdir = TempDir()
self.root_path = self._tmpdir.name
self._adapter = PixelatedDockerAdapter()
开发者ID:mmircea16，项目名称:pixelated-dispatcher，代码行数:4，代码来源:docker_provider_test.py
示例12: setUp
def setUp(self):
self._tmpdir = TempDir()
self.ssl_request = requests.Session()
self.ssl_request.mount('https://', EnforceTLSv1Adapter())
开发者ID:mmircea16，项目名称:pixelated-dispatcher，代码行数:4，代码来源:smoke_test.py
示例13: setUp
def setUp(self):
self._tmpdir = TempDir()
self.root_path = self._tmpdir.name
self.users = Users(self.root_path)
开发者ID:vivekbsable，项目名称:pixelated-dispatcher，代码行数:4，代码来源:users_test.py
示例14: AppTestClient
class AppTestClient(object):
INDEX_KEY = (
"\xde3?\x87\xff\xd9\xd3\x14\xf0\xa7>\x1f%C{\x16.\\\xae\x8c\x13\xa7\xfb\x04\xd4]+\x8d_\xed\xd1\x8d\x0bI"
"\x8a\x0e\xa4tm\xab\xbf\xb4\xa5\x99\x00d\xd5w\x9f\x18\xbc\x1d\xd4_W\xd2\xb6\xe8H\x83\x1b\xd8\x9d\xad"
)
ACCOUNT = "test"
MAIL_ADDRESS = "[email protected]"
def _initialize(self):
self._tmp_dir = TempDir()
self.accounts = {}
@defer.inlineCallbacks
def start_client(self, mode=UserAgentMode(is_single_user=True)):
self._initialize()
self._mode = mode
self._test_account = AppTestAccount(self.ACCOUNT, self._tmp_dir.name)
yield self._test_account.start()
self.cleanup = lambda: self._tmp_dir.dissolve()
# copy fields for single user tests
self.soledad = self._test_account.soledad
self.search_engine = self._test_account.search_engine
self.keymanager = self._test_account.keymanager
self.mail_sender = self._test_account.mail_sender
self.mail_store = self._test_account.mail_store
self.attachment_store = self._test_account.attachment_store
self.draft_service = self._test_account.draft_service
self.leap_session = self._test_account.leap_session
self.feedback_service = self._test_account.feedback_service
self.mail_service = self._test_account.mail_service
self.account = self._test_account.account
if mode.is_single_user:
self.service_factory = SingleUserServicesFactory(mode)
services = self._test_account.services
self.service_factory.add_session("someuserid", services)
self.resource = RootResource(self.service_factory)
self.resource.initialize()
else:
self.service_factory = StubServicesFactory(self.accounts, mode)
provider = mock()
provider.config = LeapConfig(self._tmp_dir.name)
self.resource = set_up_protected_resources(
RootResource(self.service_factory), provider, self.service_factory, checker=StubSRPChecker(provider)
)
@defer.inlineCallbacks
def create_user(self, account_name):
if self._mode.is_single_user:
raise Exception("Not supported in single user mode")
account = AppTestAccount(account_name, self._tmp_dir.name)
yield account.start()
self.accounts[account_name] = account
def _render(self, request, as_json=True):
def get_str(_str):
return json.loads(_str) if as_json else _str
def get_request_written_data(_=None):
written_data = request.getWrittenData()
if written_data:
return get_str(written_data)
resource = getChildForRequest(self.resource, request)
result = resource.render(request)
if isinstance(result, basestring):
return get_str(result), request
# result is NOT_DONE_YET
d = succeed(request) if request.finished else request.notifyFinish()
d.addCallback(get_request_written_data)
return d, request
def listenTCP(self, port=4567, host="127.0.0.1"):
reactor.listenTCP(port, PixelatedSite(self.resource), interface=host)
def run_on_a_thread(self, logfile="/tmp/app_test_client.log", port=4567, host="127.0.0.1"):
def _start():
self.listenTCP(port, host)
reactor.run()
process = multiprocessing.Process(target=_start)
process.start()
time.sleep(1)
return lambda: process.terminate()
def stop(self):
reactor.stop()
def get(self, path, get_args="", as_json=True, ajax=True, csrf="token"):
request = request_mock(path, ajax=ajax, csrf=csrf)
request.args = get_args
#.........这里部分代码省略.........
开发者ID:akjones，项目名称:pixelated-user-agent，代码行数:101，代码来源:app_test_client.py
示例15: setUp
def setUp(self):
self.tempdir = TempDir()
self.user_home = self.tempdir.name
开发者ID:carriercomm，项目名称:pixelated-user-agent，代码行数:3，代码来源:test_search.py
示例16: setUp
def setUp(self):
self.tempdir = TempDir()
self.agent_home = self.tempdir.name
开发者ID:amukiza，项目名称:pixelated-user-agent，代码行数:3，代码来源:test_search.py
示例17: RESTfulServerTest
class RESTfulServerTest(unittest.TestCase):
mock_provider = None
ssl_config = None
server = None
@classmethod
def setUpClass(cls):
RESTfulServerTest.mock_provider = MagicMock(spec=Provider)
RESTfulServerTest.mock_users = MagicMock(spec=Users)
RESTfulServerTest.mock_authenticator = MagicMock(spec=Authenticator)
RESTfulServerTest.ssl_config = SSLConfig(certfile(),
keyfile())
RESTfulServerTest.server = RESTfulServer(RESTfulServerTest.ssl_config, RESTfulServerTest.mock_users, RESTfulServerTest.mock_authenticator, RESTfulServerTest.mock_provider)
RESTfulServerTest.server.server_forever_in_backgroud()
time.sleep(1) # let it get up to speed
@classmethod
def tearDownClass(cls):
RESTfulServerTest.server.shutdown()
print 'Stopped test server'
def setUp(self):
self.mock_provider = RESTfulServerTest.mock_provider
self.mock_provider.reset_mock()
self.mock_users.reset_mock()
self.mock_authenticator.reset_mock()
self.ssl_request = requests.Session()
self.ssl_request.mount('https://', EnforceTLSv1Adapter())
self._tmpdir = TempDir()
self._root_path = self._tmpdir.name
def tearDown(self):
self._tmpdir.dissolve()
def get(self, url):
return self.ssl_request.get(url, verify=cafile())
def put(self, url, data=None):
if data:
data = json.dumps(data)
return self.ssl_request.put(url, data=data, headers={'content-type': 'application/json'}, verify=cafile())
def post(self, url, data=None):
if data:
data = json.dumps(data)
return self.ssl_request.post(url, data=data, headers={'content-type': 'application/json'}, verify=cafile())
def delete(self, url):
return self.ssl_request.delete(url, verify=cafile())
def assertSuccessJson(self, dict, response):
self.assertEqual(200, response.status_code)
self.assertEquals('application/json', response.headers['content-type'])
self.assertEqual(dict, response.json())
def assertInternalError(self, response):
self.assertEqual(500, response.status_code)
def test_list_empty_agents(self):
# given
self.mock_provider.status.return_value = {'state': 'stopped'}
self.mock_users.list.return_value = []
# when
r = self.get('https://localhost:4443/agents')
# then
self.assertSuccessJson({'agents': []}, r)
def test_list_agents(self):
# given
self.mock_provider.status.return_value = {'state': 'stopped'}
self.mock_provider.status.side_effect = None
self.mock_users.list.return_value = ['first', 'second']
# when
r = self.get('https://localhost:4443/agents')
# then
self.assertSuccessJson(
{'agents': [
{'name': 'first', 'state': 'stopped', 'uri': 'http://localhost:4443/agents/first'},
{'name': 'second', 'state': 'stopped', 'uri': 'http://localhost:4443/agents/second'}
]}, r)
def test_add_agent(self):
# given
self.mock_provider.status.return_value = {'state': 'stopped'}
payload = {'name': 'first', 'password': 'some password'}
# when
r = self.post('https://localhost:4443/agents', data=payload)
# then
#.........这里部分代码省略.........
开发者ID:Meistache，项目名称:pixelated-dispatcher，代码行数:101，代码来源:manager_test.py
示例18: ForkProviderTest
class ForkProviderTest(unittest.TestCase):
def setUp(self):
self._tmpdir = TempDir()
self.root_path = self._tmpdir.name
self.runner = MagicMock(spec=Adapter)
self.provider = ForkProvider(self.runner)
self.provider.initialize()
def tearDown(self):
self._tmpdir.dissolve()
def test_instances_can_be_removed(self):
user_config = UserConfig('test', join(self.root_path, 'test'))
os.makedirs(join(user_config.path, 'data'))
self.provider.remove(user_config)
def test_that_instance_can_be_started_and_gets_initialized(self):
self.provider.start(self._user_config('test'))
self.runner.initialize.assert_called_with('test')
self.runner.start.assert_called_with('test')
def test_that_instance_cannot_be_started_twice(self):
user_config = self._user_config('test')
self.provider.start(user_config)
self.assertRaises(InstanceAlreadyRunningError, self.provider.start, user_config)
def test_that_running_instances_are_in_runnig_list(self):
self._init_runner_memory_usage()
for name in ['one', 'two', 'three']:
self.provider.start(self._user_config(name))
self.assertEqual({'one', 'two', 'three'}, set(self.provider.list_running()))
def test_that_non_existing_instance_cannot_be_stopped(self):
self.assertRaises(InstanceNotRunningError, self.provider.stop, 'test')
def test_that_non_started_instance_cannot_be_stopped(self):
self.assertRaises(InstanceNotRunningError, self.provider.stop, UserConfig('not-started', None))
def test_that_running_instance_can_be_stopped(self):
process = MagicMock(spec=ForkedProcess)
self.runner.start.return_value = process
user_config = self._user_config('test')
self.provider.start(user_config)
self.provider.stop(user_config.username)
process.terminate.assert_called_once_with()
def test_that_instance_cannot_be_stopped_twice(self):
self.provider.start(self._user_config('test'))
self.provider.stop('test')
self.assertRaises(InstanceNotRunningError, self.provider.stop, 'test')
def test_that_status_returns_current_port(self):
process = MagicMock(spec=ForkedProcess(None, 1234))
process.port = 1234
self.runner.start.return_value = process
self.provider.start(self._user_config('test'))
status = self.provider.status('test')
self.assertEqual({'port': 1234, 'state': 'running'}, status)
def assert_config_file(self, filename, name, hashed_password, salt):
with open(filename, 'r') as file:
content = file.read()
self.assertEqual('[agent]\nname = %s\nhashed_password = %s\nsalt = %s\n\n' % (name, hashed_password, salt),
content)
def test_memory_usage_zero_if_no_processes(self):
self.assertEqual({'total_usage': 0, 'average_usage': 0, 'agents': []}, self.provider.memory_usage())
def test_memory_usage_with_process(self):
# given
process = MagicMock(spec=ForkedProcess(None, 1234))
process.port = 1234
process.memory_usage.return_value = 1024
self.runner.start.return_value = process
self.provider.start(self._user_config('test'))
# when
usage = self.provider.memory_usage()
# then
self.assertEqual({'total_usage': 1024,
'average_usage': 1024,
'agents': [
{'name': 'test', 'memory_usage': 1024}
]}, usage)
@patch('pixelated.provider.fork.psutil.virtual_memory')
def test_that_instance_cannot_be_started_with_too_little_memory_left(self, vm_mock):
# given
svmem = namedtuple('svmem', ['free'])
free_memory = 1024 * 1024
vm_mock.return_value = svmem(free_memory)
#.........这里部分代码省略.........
开发者ID:Meistache，项目名称:pixelated-dispatcher，代码行数:101，代码来源:provider_test.py
示例19: TestDatabase
class TestDatabase(unittest.TestCase):
domain = 'domain'
username = 'username'
plainpassword = 'password'
cipherpassword = '69e00ec7020a4d22'.decode('hex')
pin = '1234'
key = hashlib.sha256(domain + username + pin).digest()
iv = 'a3206f4194d1d7a252a9fe24b7b063b9'.decode('hex')
def setUp(self):
self.d = TempDir()
self.keyfile_path = os.path.join(self.d.name, 'keyfile')
self.database_path = os.path.join(self.d.name, 'database')
self.keyfile = K.create(
self.keyfile_path,
self.database_path)
D.create(self.keyfile)
def tearDown(self):
self.d.dissolve()
@raises(IOError)
def test_create_database_exists(self):
D.create(self.keyfile)
@raises(AttributeError)
def test_enter_missing_key(self):
del self.keyfile.key
with D(self.keyfile):
pass
@raises(ValueError)
def test_enter_invalid_key(self):
self.keyfile.key = 'not a key'
with D(self.keyfile):
pass
@raises(ValueError)
def test_enter_incorrect_key(self):
self.keyfile.key = '1b660492d98c54412d3d91818de5a2ae0b3110850a12010768b80fb277f55aa5'.decode('hex')
with D(self.keyfile):
pass
@raises(AttributeError)
def test_enter_missing_iv(self):
del self.keyfile.iv
with D(self.keyfile):
pass
@raises(ValueError)
def test_enter_invalid_iv(self):
self.keyfile.iv = 'not an iv'
with D(self.keyfile):
pass
@raises(ValueError)
def test_enter_incorrect_iv(self):
self.keyfile.iv = 'a059d464b93397a2a98e8e1f00b596c7'.decode('hex')
with D(self.keyfile):
pass
@raises(AttributeError)
def test_enter_missing_path(self):
del self.keyfile.database_path
with D(self.keyfile):
pass
@raises(IOError)
def test_enter_unreadable_path(self):
self.keyfile.database_path = '/nopermission'
with D(self.keyfile):
pass
@raises(ValueError)
def test_enter_not_a_database_path(self):
self.keyfile.database_path = os.path.join(self.d.name, 'database')
with open(self.keyfile.database_path, 'wb') as f:
f.write(Random.new().read(10))
with D(self.keyfile):
pass
def test_exit_saves_changes(self):
c = C(self.domain, self.username, cipherpassword=self.cipherpassword, iv=self.iv)
with D(self.keyfile) as d:
d.add(c, self.pin)
with D(self.keyfile) as d:
self.assertEqual(
[c],
d.find(self.domain, self.username))
def test_add(self):
c1 = C(self.domain, self.username, cipherpassword=self.cipherpassword, iv=self.iv)
with D(self.keyfile) as d:
d.add(c1, self.pin)
c2 = d.find(self.domain, self.username)[0]
print c1
print c2
self.assertEqual(self.plainpassword, c2.unlock(self.pin))
#.........这里部分代码省略.........
开发者ID:WilliamMayor，项目名称:pinscher，代码行数:101，代码来源:test_database.py
示例20: SearchEngineTest
class SearchEngineTest(unittest.TestCase):
def setUp(self):
self.tempdir = TempDir()
self.user_home = self.tempdir.name
def tearDown(self):
self.tempdir.dissolve()
def test_headers_encoding(self):
# given
se = SearchEngine(INDEX_KEY, self.user_home)
headers = {
'From': '[email protected]',
'To': '=?utf-8?b?IsOEw7zDtiDDlsO8w6QiIDxmb2xrZXJAcGl4ZWxhdGVkLXByb2plY3Qub3Jn?=\n =?utf-8?b?PiwgRsO2bGtlciA8Zm9sa2VyQHBpeGVsYXRlZC1wcm9qZWN0Lm9yZz4=?=',
'Cc': '=?utf-8?b?IsOEw7zDtiDDlsO8w6QiIDxmb2xrZXJAcGl4ZWxhdGVkLXByb2plY3Qub3Jn?=\n =?utf-8?b?PiwgRsO2bGtlciA8Zm9sa2VyQHBpeGVsYXRlZC1wcm9qZWN0Lm9yZz4=?=',
'Subject': 'Some test mail',
}
# when
se.index_mail(LeapMail('mailid', 'INBOX', headers=headers)) # test_helper.pixelated_mail(extra_headers=headers, chash='mailid'))
result = se.search('folker')
self.assertEqual((['mailid'], 1), result)
def test_contents_encoding_accents(self):
# given
se = SearchEngine(INDEX_KEY, self.user_home)
headers = {
'From': '[email protected]',
'To': '=?utf-8?b?IsOEw7zDtiDDlsO8w6QiIDxmb2xrZXJAcGl4ZWxhdGVkLXByb2plY3Qub3Jn?=\n =?utf-8?b?PiwgRsO2bGtlciA8Zm9sa2VyQHBpeGVsYXRlZC1wcm9qZWN0Lm9yZz4=?=',
'Cc': '=?utf-8?b?IsOEw7zDtiDDlsO8w6QiIDxmb2xrZXJAcGl4ZWxhdGVkLXByb2plY3Qub3Jn?=\n =?utf-8?b?PiwgRsO2bGtlciA8Zm9sa2VyQHBpeGVsYXRlZC1wcm9qZWN0Lm9yZz4=?=',
'Subject': 'Some test mail',
}
body = "When doing the search, it's not possible to find words with graphical accents, e.g.: 'coração', 'é', 'Fièvre', La Pluie d'été, 'não'."
# when
se.index_mail(LeapMail('mailid', 'INBOX', headers=headers, body=body)) # test_helper.pixelated_mail(extra_headers=headers, chash='mailid'))
result = se.search(u"'coração', 'é',")
self.assertEqual((['mailid'], 1), result)
result = se.search(u"Fièvre")
self.assertEqual((['mailid'], 1), result)
result = se.search(u"été")
self.assertEqual((['mailid'], 1), result)
def test_contents_encoding_special_characters(self):
# given
se = SearchEngine(INDEX_KEY, self.user_home)
headers = {
'From': '[email protected]',
'To': '=?utf-8?b?IsOEw7zDtiDDlsO8w6QiIDxmb2xrZXJAcGl4ZWxhdGVkLXByb2plY3Qub3Jn?=\n =?utf-8?b?PiwgRsO2bGtlciA8Zm9sa2VyQHBpeGVsYXRlZC1wcm9qZWN0Lm9yZz4=?=',
'Cc': '=?utf-8?b?IsOEw7zDtiDDlsO8w6QiIDxmb2xrZXJAcGl4ZWxhdGVkLXByb2plY3Qub3Jn?=\n =?utf-8?b?PiwgRsO2bGtlciA8Zm9sa2VyQHBpeGVsYXRlZC1wcm9qZWN0Lm9yZz4=?=',
'Subject': 'Some test mail',
}
body = "When doing the search, 您好 أهلا"
# when
se.index_mail(LeapMail('mailid', 'INBOX', headers=headers, body=body)) # test_helper.pixelated_mail(extra_headers=headers, chash='mailid'))
result = se.search(u"您好")
self.assertEqual((['mailid'], 1), result)
result = se.search(u"أهلا")
self.assertEqual((['mailid'], 1), result)
开发者ID:carriercomm，项目名称:pixelated-user-agent，代码行数:72，代码来源:test_search.py
注：本文中的tempdir.TempDir类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论