def test_SADParser(self):
sad = parsers.SADParser.parse("sdr/dom/waveforms/CommandWrapperWithPropertyOverride/CommandWrapper.sad.xml")
self.assertEqual(sad.get_id(), "DCE:d206ab51-6342-4976-bac3-55e6902f3489")
self.assertEqual(sad.get_name(), "CommandWrapperWithPropertyOverride")
self.assertEqual(len(sad.componentfiles.get_componentfile()), 1)
self.assertEqual(len(sad.partitioning.get_componentplacement()), 1)
self.assertEqual(sad.partitioning.get_componentplacement()[0].componentfileref.refid, "CommandWrapper_592b8bd6-b011-4468-9417-705af45e907b")
self.assertEqual(sad.partitioning.get_componentplacement()[0].get_componentinstantiation()[0].id_, "DCE:8c129782-a6a4-4095-8212-757f01de0c09")
self.assertEqual(sad.partitioning.get_componentplacement()[0].get_componentinstantiation()[0].get_usagename(), "CommandWrapper1")
self.assertEqual(sad.partitioning.get_componentplacement()[0].get_componentinstantiation()[0].componentproperties.get_simpleref()[0].refid, "DCE:a4e7b230-1d17-4a86-aeff-ddc6ea3df26e")
self.assertEqual(sad.partitioning.get_componentplacement()[0].get_componentinstantiation()[0].componentproperties.get_simpleref()[0].value, "/bin/date")
# Verify that we can write the output and still be DTD valid
tmpfile = tempfile.mktemp()
try:
tmp = open(tmpfile, "w")
sad.export(tmp, 0)
tmp.close()
status = self._xmllint(tmpfile, "SAD")
self.assertEqual(status, 0, "Python parser did not emit DTD compliant XML")
finally:
try:
os.remove(tmpfile)
except OSError:
pass
python类mktemp()的实例源码
def test_get_swift_hash_env(self, mock_config, mock_service_name):
mock_config.return_value = None
mock_service_name.return_value = "testsvc"
tmpfile = tempfile.mktemp()
swift_context.SWIFT_HASH_FILE = tmpfile
with mock.patch('lib.swift_context.os.environ.get') as mock_env_get:
mock_env_get.return_value = str(uuid.uuid4())
hash_ = swift_context.get_swift_hash()
mock_env_get.assert_has_calls([
mock.call('JUJU_MODEL_UUID'),
mock.call('JUJU_ENV_UUID',
mock_env_get.return_value)
])
with open(tmpfile, 'r') as fd:
self.assertEqual(hash_, fd.read())
self.assertTrue(mock_config.called)
def generate_pdf(card):
"""
Make a PDF from a card
:param card: dict from fetcher.py
:return: Binary PDF buffer
"""
from eclaire.base import SPECIAL_LABELS
pdf = FPDF('L', 'mm', (62, 140))
pdf.set_margins(2.8, 2.8, 2.8)
pdf.set_auto_page_break(False, margin=0)
pdf.add_page()
font = pkg_resources.resource_filename('eclaire', 'font/Clairifont.ttf')
pdf.add_font('Clairifont', fname=font, uni=True)
pdf.set_font('Clairifont', size=48)
pdf.multi_cell(0, 18, txt=card.name.upper(), align='L')
qrcode = generate_qr_code(card.url)
qrcode_file = mktemp(suffix='.png', prefix='trello_qr_')
qrcode.save(qrcode_file)
pdf.image(qrcode_file, 118, 35, 20, 20)
os.unlink(qrcode_file)
# May we never speak of this again.
pdf.set_fill_color(255, 255, 255)
pdf.rect(0, 55, 140, 20, 'F')
pdf.set_font('Clairifont', '', 16)
pdf.set_y(-4)
labels = ', '.join([label.name for label in card.labels
if label.name not in SPECIAL_LABELS])
pdf.multi_cell(0, 0, labels, 0, 'R')
return pdf.output(dest='S')
def test_exit_crash():
# For each Widget subclass, run a simple python script that creates an
# instance and then shuts down. The intent is to check for segmentation
# faults when each script exits.
tmp = tempfile.mktemp(".py")
path = os.path.dirname(pg.__file__)
initArgs = {
'CheckTable': "[]",
'ProgressDialog': '"msg"',
'VerticalLabel': '"msg"',
}
for name in dir(pg):
obj = getattr(pg, name)
if not isinstance(obj, type) or not issubclass(obj, pg.QtGui.QWidget):
continue
print(name)
argstr = initArgs.get(name, "")
open(tmp, 'w').write(code.format(path=path, classname=name, args=argstr))
proc = subprocess.Popen([sys.executable, tmp])
assert proc.wait() == 0
os.remove(tmp)
def test_exit_crash():
# For each Widget subclass, run a simple python script that creates an
# instance and then shuts down. The intent is to check for segmentation
# faults when each script exits.
tmp = tempfile.mktemp(".py")
path = os.path.dirname(pg.__file__)
initArgs = {
'CheckTable': "[]",
'ProgressDialog': '"msg"',
'VerticalLabel': '"msg"',
}
for name in dir(pg):
obj = getattr(pg, name)
if not isinstance(obj, type) or not issubclass(obj, pg.QtGui.QWidget):
continue
print(name)
argstr = initArgs.get(name, "")
open(tmp, 'w').write(code.format(path=path, classname=name, args=argstr))
proc = subprocess.Popen([sys.executable, tmp])
assert proc.wait() == 0
os.remove(tmp)
def _smcra_to_str(self, smcra, temp_dir='/tmp/'):
"""
WHATIF's input are PDB format files.
Converts a SMCRA object to a PDB formatted string.
"""
temp_path = tempfile.mktemp( '.pdb', dir=temp_dir )
io = PDBIO()
io.set_structure(smcra)
io.save(temp_path)
f = open(temp_path, 'r')
string = f.read()
f.close()
os.remove(temp_path)
return string
def test_binary(self, enable_randomness=True, times=1, timeout=15):
"""
Test the binary generated
"""
# dump the binary code
pov_binary_filename = tempfile.mktemp(dir='/tmp', prefix='rex-pov-')
self.dump_binary(filename=pov_binary_filename)
os.chmod(pov_binary_filename, 0755)
pov_tester = CGCPovSimulator()
result = pov_tester.test_binary_pov(
pov_binary_filename,
self.crash.binary,
enable_randomness=enable_randomness,
timeout=timeout,
times=times)
# remove the generated pov
os.remove(pov_binary_filename)
return result
def generate_report(self, register_setters, leakers):
stat_name = tempfile.mktemp(dir=".", prefix='rex-results-')
l.info("exploitation report being written to '%s'", stat_name)
f = open(stat_name, 'w')
f.write("Binary %s:\n" % os.path.basename(self.crash.project.filename))
f.write("Register setting exploits:\n")
for register_setter in register_setters:
f.write("\t%s\n" % str(register_setter))
f.write("\n")
f.write("Leaker exploits:\n")
for leaker in leakers:
f.write("\t%s\n" % str(leaker))
f.close()
def start_docker_compose(clients=1):
inline_logs = conftest.inline_logs
docker_compose_cmd("up -d")
if clients > 1:
docker_compose_cmd("scale mender-client=%d" % clients)
if inline_logs:
docker_compose_cmd("logs -f &")
else:
tfile = tempfile.mktemp("mender_testing")
docker_compose_cmd("logs -f --no-color > %s 2>&1 &" % tfile)
logger.info("docker-compose log file stored here: %s" % tfile)
log_files.append(tfile)
ssh_is_opened()
common.set_setup_type(common.ST_OneClient)
def tearDown(self):
"""
Clean up any files or directories created using L{TestCase.mktemp}.
Subclasses must invoke this method if they override it or the
cleanup will not occur.
"""
if self._temporaryFiles is not None:
for temp in self._temporaryFiles:
if os.path.isdir(temp):
shutil.rmtree(temp)
elif os.path.exists(temp):
os.unlink(temp)
try:
_exception_from_error_queue()
except Error, e:
if e.args != ([],):
self.fail("Left over errors in OpenSSL error queue: " + repr(e))
def _getBatchOutput(self, f):
fn = tempfile.mktemp()
open(fn, 'w').write(f)
port = self.server.getHost().port
cmds = ('-p %i -l testuser '
'-K unix '
'-a '
'-v -b %s 127.0.0.1') % (port, fn)
cmds = test_conch._makeArgs(cmds.split(), mod='cftp')[1:]
log.msg('running %s %s' % (sys.executable, cmds))
env = os.environ.copy()
env['PYTHONPATH'] = os.pathsep.join(sys.path)
self.server.factory.expectedLoseConnection = 1
d = getProcessOutputAndValue(sys.executable, cmds, env=env)
def _cleanup(res):
os.remove(fn)
return res
d.addCallback(lambda res: res[0])
d.addBoth(_cleanup)
return d
def loopbackUNIX(server, client, noisy=True):
"""Run session between server and client protocol instances over UNIX socket."""
path = tempfile.mktemp()
from twisted.internet import reactor
f = policies.WrappingFactory(protocol.Factory())
serverWrapper = _FireOnClose(f, server)
f.noisy = noisy
f.buildProtocol = lambda addr: serverWrapper
serverPort = reactor.listenUNIX(path, f)
clientF = LoopbackClientFactory(client)
clientF.noisy = noisy
reactor.connectUNIX(path, clientF)
d = clientF.deferred
d.addCallback(lambda x: serverWrapper.deferred)
d.addCallback(lambda x: serverPort.stopListening())
return d
def get_all_species(self):
import tempfile
outfile = tempfile.mktemp() + '.txt.gz'
try:
self.logger.info('Downloading "species.txt.gz"...')
out_f = open(outfile, 'wb')
ftp = FTP(self.__class__.ENSEMBL_FTP_HOST)
ftp.login()
species_file = '/pub/release-%s/mysql/ensembl_production_%s/species.txt.gz' % (self.release, self.release)
ftp.retrbinary("RETR " + species_file, out_f.write)
out_f.close()
self.logger.info('Done.')
#load saved file
self.logger.info('Parsing "species.txt.gz"...')
species_li = tab2list(outfile, (1, 2, 7), header=0) # db_name,common_name,taxid
species_li = [x[:-1] + [is_int(x[-1]) and int(x[-1]) or None] for x in species_li]
# as of ensembl 87, there are also mouse strains. keep only the "original" one
species_li = [s for s in species_li if not s[0].startswith("mus_musculus_")]
self.logger.info('Done.')
finally:
os.remove(outfile)
pass
return species_li
def __init__(self,
in_file,
exception_handler,
bug_handler,
copy = None,
run_level = 1,
):
self.__file = in_file
self.__bug_handler = bug_handler
self.__copy = copy
self.__run_level = run_level
self.__write_to = tempfile.mktemp()
self.__exception_handler = exception_handler
self.__bug_handler = bug_handler
self.__state = 'outside'
self.__utf_exp = re.compile(r'&#x(.*?);')
def __init__(self,
in_file,
bug_handler,
out_file,
copy = None,
orig_file = None,
run_level = 1,
):
self.__file = in_file
self.__bug_handler = bug_handler
self.__copy = copy
self.__run_level = run_level
self.__write_to = tempfile.mktemp()
self.__bracket_count = 0
self.__ob_count = 0
self.__cb_count = 0
self.__pict_count = 0
self.__in_pict = 0
self.__already_found_pict = 0
self.__orig_file = orig_file
self.__initiate_pict_dict()
self.__out_file = out_file
# this is left over
self.__no_ask = 1
def __init__(self,
in_file ,
bug_handler,
copy = None,
run_level = 1,
):
self.__file = in_file
self.__bug_handler = bug_handler
self.__copy = copy
self.__write_to = tempfile.mktemp()
self.__bracket_count=0
self.__ob_count = 0
self.__cb_count = 0
self.__after_asterisk = 0
self.__delete = 0
self.__initiate_allow()
self.__ob = 0
self.__write_cb = 0
self.__run_level = run_level
self.__found_delete = 0
self.__list = 0
def __test_bad(self):
paths = os.listdir(self.__invalid_dir)
for path in paths:
path = os.path.join(self.__invalid_dir, path)
filename, ext = os.path.splitext(path)
if ext != '.rtf':
continue
if os.path.isdir(path):
continue
new_filename = tempfile.mktemp()
out_path = os.path.join(self.__out_dir, new_filename)
status, msg = self.__run_script(in_file = path, out_file = out_path)
if not status:
info = [path, 'Should have produced an error']
self.__error.append(info)
try:
os.remove(out_path)
except OSError:
pass
def test_smrt_archiver_lasttime():
tmpfile = tempfile.mktemp()
archiver = Archiver(dbfile=tmpfile)
rule = 'test/smrt/rules/archiver.yml'
feed = 'lasttime'
with Smrt(REMOTE_ADDR, 1234, client='stdout', archiver=archiver) as s:
assert type(s) is Smrt
for r, f in s.load_feeds(rule, feed=feed):
x = list(s.process(r, f))
assert len(x) > 0
f = {i.indicator: i.__dict__() for i in x}
assert f['216.243.31.2']['lasttime'] == '2016-03-23T20:22:27.000000Z'
with Smrt(REMOTE_ADDR, 1234, client='stdout', archiver=archiver) as s:
assert type(s) is Smrt
for r, f in s.load_feeds(rule, feed=feed):
x = list(s.process(r, f))
assert len(x) == 0
def test_smrt_archiver_firsttime():
tmpfile = tempfile.mktemp()
archiver = Archiver(dbfile=tmpfile)
rule = 'test/smrt/rules/archiver.yml'
feed = 'firsttime'
with Smrt(REMOTE_ADDR, 1234, client='stdout', archiver=archiver) as s:
assert type(s) is Smrt
for r, f in s.load_feeds(rule, feed=feed):
x = list(s.process(r, f))
assert len(x) > 0
f = {i.indicator: i.__dict__() for i in x}
assert f['216.243.31.2']['lasttime'] == '2016-03-23T20:22:27.000000Z'
with Smrt(REMOTE_ADDR, 1234, client='stdout', archiver=archiver) as s:
assert type(s) is Smrt
for r, f in s.load_feeds(rule, feed=feed):
x = list(s.process(r, f))
assert len(x) == 0
def test_smrt_archiver_both():
tmpfile = tempfile.mktemp()
archiver = Archiver(dbfile=tmpfile)
rule = 'test/smrt/rules/archiver.yml'
feed = 'both'
with Smrt(REMOTE_ADDR, 1234, client='stdout', archiver=archiver) as s:
assert type(s) is Smrt
for r, f in s.load_feeds(rule, feed=feed):
x = list(s.process(r, f))
assert len(x) > 0
f = {i.indicator: i.__dict__() for i in x}
assert f['216.243.31.2']['lasttime'] == '2016-03-23T20:22:27.000000Z'
with Smrt(REMOTE_ADDR, 1234, client='stdout', archiver=archiver) as s:
assert type(s) is Smrt
for r, f in s.load_feeds(rule, feed=feed):
x = list(s.process(r, f))
assert len(x) == 0
def test_smrt_archiver_neither():
tmpfile = tempfile.mktemp()
archiver = Archiver(dbfile=tmpfile)
rule = 'test/smrt/rules/archiver.yml'
feed = 'neither'
with Smrt(REMOTE_ADDR, 1234, client='stdout', archiver=archiver) as s:
assert type(s) is Smrt
for r, f in s.load_feeds(rule, feed=feed):
x = list(s.process(r, f))
assert len(x) > 0
f = {i.indicator: i.__dict__() for i in x}
assert f['216.243.31.2'].get('lasttime') is None
with Smrt(REMOTE_ADDR, 1234, client='stdout', archiver=archiver) as s:
assert type(s) is Smrt
for r, f in s.load_feeds(rule, feed=feed):
x = list(s.process(r, f))
assert len(x) == 0
def test_smrt_archiver_lasttime_clear():
tmpfile = tempfile.mktemp()
archiver = Archiver(dbfile=tmpfile)
rule = 'test/smrt/rules/archiver.yml'
feed = 'lasttime'
with Smrt(REMOTE_ADDR, 1234, client='stdout', archiver=archiver) as s:
assert type(s) is Smrt
for r, f in s.load_feeds(rule, feed=feed):
x = list(s.process(r, f))
assert len(x) > 0
f = {i.indicator: i.__dict__() for i in x}
assert f['216.243.31.2']['lasttime'] == '2016-03-23T20:22:27.000000Z'
archiver.clear_memcache()
with Smrt(REMOTE_ADDR, 1234, client='stdout', archiver=archiver) as s:
assert type(s) is Smrt
for r, f in s.load_feeds(rule, feed=feed):
x = list(s.process(r, f))
assert len(x) == 0
def get_local_filename(self):
""" get_local_filename()
If the filename is an existing file on this filesystem, return
that. Otherwise a temporary file is created on the local file
system which can be used by the format to read from or write to.
"""
if self._uri_type == URI_FILENAME:
return self._filename
else:
# Get filename
ext = os.path.splitext(self._filename)[1]
self._filename_local = tempfile.mktemp(ext, 'imageio_')
# Write stuff to it?
if self.mode[0] == 'r':
with open(self._filename_local, 'wb') as file:
shutil.copyfileobj(self.get_file(), file)
return self._filename_local
def test_version_2_0_memmap():
# requires more than 2 byte for header
dt = [(("%d" % i) * 100, float) for i in range(500)]
d = np.ones(1000, dtype=dt)
tf = tempfile.mktemp('', 'mmap', dir=tempdir)
# 1.0 requested but data cannot be saved this way
assert_raises(ValueError, format.open_memmap, tf, mode='w+', dtype=d.dtype,
shape=d.shape, version=(1, 0))
ma = format.open_memmap(tf, mode='w+', dtype=d.dtype,
shape=d.shape, version=(2, 0))
ma[...] = d
del ma
with warnings.catch_warnings(record=True) as w:
warnings.filterwarnings('always', '', UserWarning)
ma = format.open_memmap(tf, mode='w+', dtype=d.dtype,
shape=d.shape, version=None)
assert_(w[0].category is UserWarning)
ma[...] = d
del ma
ma = format.open_memmap(tf, mode='r')
assert_array_equal(ma, d)
def testAll(self):
output_name = tempfile.mktemp("-pycom-test")
cmd = "cscript //nologo testxslt.js doesnt_matter.xml testxslt.xsl " + output_name
win32com.test.util.ExecuteShellCommand(cmd, self)
try:
f=open(output_name)
try:
got = f.read()
if got != expected_output:
print "ERROR: XSLT expected output of %r" % (expected_output,)
print "but got %r" % (got,)
finally:
f.close()
finally:
try:
os.unlink(output_name)
except os.error:
pass
def _SetCurrent(self, image):
filename = '%d.iso' % (image['timestamp'])
path = os.path.join(self._image_dir, filename)
current_path = os.path.join(self._image_dir, 'current')
try:
link = os.readlink(current_path)
link_path = os.path.join(self._image_dir, link)
if link_path == path:
return
except FileNotFoundError:
pass
print('Changing current link to:', filename, flush=True)
temp_path = tempfile.mktemp(dir=self._image_dir)
os.symlink(filename, temp_path)
os.rename(temp_path, current_path)
def _write_styles_file(self):
self.styles_xml = tempfile.mktemp()
try:
self.f = open(self.styles_xml,"wb")
except IOError as msg:
errmsg = "%s\n%s" % (_("Could not create %s") % self.styles_xml, msg)
raise ReportError(errmsg)
except:
raise ReportError(_("Could not create %s") % self.styles_xml)
self.f = open(self.styles_xml,"w")
self.f.write('<?xml version="1.0" encoding="UTF-8"?>\n')
self.f.write(
'<office:document-styles ' +
_XMLNS +
'office:version="1.0"> '
)
self.f.write(_STYLES_FONTS)
self.f.write(_STYLES_STYLES)
self.f.write(_STYLES_AUTOMATIC)
self.f.write(_STYLES_MASTER)
self.f.write('</office:document-styles>\n')
self.f.close()
def _write_meta_file(self):
self.meta_xml = tempfile.mktemp()
try:
self.f = open(self.meta_xml,"wb")
except IOError as msg:
errmsg = "%s\n%s" % (_("Could not create %s") % self.meta_xml, msg)
raise ReportError(errmsg)
except:
raise ReportError(_("Could not create %s") % self.meta_xml)
self.f = open(self.meta_xml,"w")
self.f.write(_META %
{'program': PROGRAM_NAME,
'version': VERSION,
'name' : self.name,
'time' : self.time,
}
)
self.f.close()
def __getstate__(self):
'''This function is neccessary for pickling'''
# Translate everything but the svm because that cannot be simply pickled.
state = {}
for key,value in self.__dict__.iteritems():
if key == '_model':
filename = tempfile.mktemp()
self._model.save(filename)
buffer = open(filename).read()
os.remove(filename)
state[key] = buffer
continue
state[key] = value
return state
def __getstate__(self):
'''This function is neccessary for pickling'''
# Translate everything but the svm because that cannot be simply pickled.
state = {}
for key,value in self.__dict__.iteritems():
if key == '_model':
filename = tempfile.mktemp()
self._model.save(filename)
buffer = open(filename).read()
os.remove(filename)
state[key] = buffer
continue
state[key] = value
return state
