我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用tempfile.mktemp()。
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
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 _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 test_resampling_value(shapefile, bins, samples): samples = (samples//bins) * bins lonlats, filename = shapefile random_filename = tempfile.mktemp() + ".shp" resampling.resample_by_magnitude(filename, random_filename, target_field='lat', bins=bins, output_samples=samples, bootstrap=True ) resampled_sf = shp.Reader(random_filename) resampled_shapefields = [f[0] for f in resampled_sf.fields[1:]] new_coords, new_val, new_othervals = \ geoio.load_shapefile(random_filename, 'lat') assert 'lat' in resampled_shapefields assert 'lon' not in resampled_shapefields assert np.all((samples, 2) == new_coords.shape) assert new_othervals == {} # only the target is retained after resampling
def test_resampling_spatial(shapefile, rows, cols, samples): tiles = rows * cols samples = (samples // tiles) * tiles lonlats, filename = shapefile random_filename = tempfile.mktemp() + ".shp" resampling.resample_spatially(filename, random_filename, target_field='lat', rows=rows, cols=cols, output_samples=samples, bootstrap=True ) resampled_sf = shp.Reader(random_filename) resampled_shapefields = [f[0] for f in resampled_sf.fields[1:]] new_coords, new_val, new_othervals = \ geoio.load_shapefile(random_filename, 'lat') assert 'lat' in resampled_shapefields assert 'lon' not in resampled_shapefields assert np.all((samples, 2) >= new_coords.shape) assert new_othervals == {} # only the target is retained after resampling
def test_do_work(self): # input_file, mask_file, output_file, resampling, extents, jpeg output_file = tempfile.mktemp(suffix='.tif') options = Options(resampling='bilinear', extents=self.extents, jpeg=True, reproject=True) crop.do_work(input_file=self.std2000_no_mask, output_file=output_file, options=options, mask_file=self.mask) # output file was created self.assertTrue(exists(output_file)) # assert jpeg was created self.assertTrue(exists( join(crop.TMPDIR, basename(output_file).split('.')[0] + '.jpg'))) os.remove(output_file)
def atomic_write(path): tmp = tempfile.mktemp( prefix='.' + os.path.basename(path), dir=os.path.dirname(path), ) try: with open(tmp, 'w') as f: yield f except BaseException: os.remove(tmp) raise else: os.rename(tmp, path) # TODO: at some point there will be so many options we'll want to make a config # object or similar instead of adding more arguments here
def screenshot(self, filename=None): """ Return: PIL.Image Raises: EnvironmentError """ tmpfile = tempfile.mktemp(prefix='atx-screencap-', suffix='.tiff') try: idevice("screenshot", "--udid", self.udid, tmpfile) except subprocess.CalledProcessError as e: sys.exit(e.message) try: image = Image.open(tmpfile) image.load() if filename: image.save(filename) return image finally: if os.path.exists(tmpfile): os.unlink(tmpfile)
def _adb_screencap(self, scale=1.0): """ capture screen with adb shell screencap """ remote_file = tempfile.mktemp(dir='/data/local/tmp/', prefix='screencap-', suffix='.png') local_file = tempfile.mktemp(prefix='atx-screencap-', suffix='.png') self.shell('screencap', '-p', remote_file) try: self.pull(remote_file, local_file) image = imutils.open_as_pillow(local_file) if scale is not None and scale != 1.0: image = image.resize([int(scale * s) for s in image.size], Image.BICUBIC) rotation = self.rotation() if rotation: method = getattr(Image, 'ROTATE_{}'.format(rotation*90)) image = image.transpose(method) return image finally: self.remove(remote_file) os.unlink(local_file)
def _adb_minicap(self, scale=1.0): """ capture screen with minicap https://github.com/openstf/minicap """ remote_file = tempfile.mktemp(dir='/data/local/tmp/', prefix='minicap-', suffix='.jpg') local_file = tempfile.mktemp(prefix='atx-minicap-', suffix='.jpg') (w, h, r) = self.display params = '{x}x{y}@{rx}x{ry}/{r}'.format(x=w, y=h, rx=int(w*scale), ry=int(h*scale), r=r*90) try: self.shell('LD_LIBRARY_PATH=/data/local/tmp', self.__minicap, '-s', '-P', params, '>', remote_file) self.pull(remote_file, local_file) image = imutils.open_as_pillow(local_file) return image finally: self.remove(remote_file) os.unlink(local_file)
def test_blocking_lock_file(self): my_file = tempfile.mktemp() lock_file = BlockingLockFile(my_file) lock_file._obtain_lock() # next one waits for the lock start = time.time() wait_time = 0.1 wait_lock = BlockingLockFile(my_file, 0.05, wait_time) self.failUnlessRaises(IOError, wait_lock._obtain_lock) elapsed = time.time() - start extra_time = 0.02 if is_win: # for Appveyor extra_time *= 6 # NOTE: Indeterministic failures here... self.assertLess(elapsed, wait_time + extra_time)
def setUp(self): self._dbfile = tempfile.mktemp()
def test_SPDParser(self): # Verify the input is valid spdPath = os.path.abspath("sdr/dom/components/CommandWrapper/CommandWrapper.spd.xml") status = self._xmllint(spdPath, "SPD") self.assertEqual(status, 0, "Input XML isn't DTD valid") spd = parsers.SPDParser.parse(spdPath) self.assertEqual(spd.get_id(), "DCE:458872f6-a316-4082-b1eb-ce5704f5c49d") self.assertEqual(spd.get_name(), "CommandWrapper") self.assertEqual(str(spd.get_author()[0].get_name()[0]), "REDHAWK test author") self.assertEqual(spd.get_propertyfile().get_type(), "PRF") self.assertEqual(spd.get_propertyfile().get_localfile().get_name(), "CommandWrapper.prf.xml") # Verify that we can write the output and still be DTD valid tmpfile = tempfile.mktemp() try: tmp = open(tmpfile, "w") spd.export(tmp, 0) tmp.close() status = self._xmllint(tmpfile, "SPD") self.assertEqual(status, 0, "Python parser did not emit DTD compliant XML") finally: try: os.remove(tmpfile) except OSError: pass
def test_PRFParser(self): prf = parsers.PRFParser.parse("sdr/dom/components/CommandWrapper/CommandWrapper.prf.xml") props = {} for property in prf.get_simple(): props[property.get_id()] = property for property in prf.get_simplesequence(): props[property.get_id()] = property self.assertEqual(props["DCE:a4e7b230-1d17-4a86-aeff-ddc6ea3df26e"].get_mode(), "readwrite") self.assertEqual(props["DCE:a4e7b230-1d17-4a86-aeff-ddc6ea3df26e"].get_name(), "command") self.assertEqual(props["DCE:a4e7b230-1d17-4a86-aeff-ddc6ea3df26e"].get_type(), "string") self.assertEqual(props["DCE:a4e7b230-1d17-4a86-aeff-ddc6ea3df26e"].get_value(), "/bin/echo") self.assertEqual(props["DCE:a4e7b230-1d17-4a86-aeff-ddc6ea3df26e"].get_kind()[0].get_kindtype(), "configure") self.assertEqual(props["DCE:5d8bfe8d-bc25-4f26-8144-248bc343aa53"].get_mode(), "readwrite") self.assertEqual(props["DCE:5d8bfe8d-bc25-4f26-8144-248bc343aa53"].get_name(), "args") self.assertEqual(props["DCE:5d8bfe8d-bc25-4f26-8144-248bc343aa53"].get_type(), "string") self.assertEqual(props["DCE:5d8bfe8d-bc25-4f26-8144-248bc343aa53"].get_values().get_value()[0], "Hello World") self.assertEqual(props["DCE:5d8bfe8d-bc25-4f26-8144-248bc343aa53"].get_kind()[0].get_kindtype(), "configure") # Verify that we can write the output and still be DTD valid tmpfile = tempfile.mktemp() try: tmp = open(tmpfile, "w") prf.export(tmp, 0) tmp.close() status = self._xmllint(tmpfile, "PRF") self.assertEqual(status, 0, "Python parser did not emit DTD compliant XML") finally: try: os.remove(tmpfile) except OSError: pass
def test_SCDParser(self): scd = parsers.SCDParser.parse("sdr/dom/components/CommandWrapper/CommandWrapper.scd.xml") self.assertEqual(scd.get_corbaversion(), "2.2") self.assertEqual(scd.get_componentrepid().get_repid(), "IDL:CF/Resource:1.0") self.assertEqual(scd.get_componenttype(), "resource") self.assertEqual(scd.get_componentfeatures().get_supportsinterface()[0].get_repid(), "IDL:CF/Resource:1.0") self.assertEqual(scd.get_componentfeatures().get_supportsinterface()[0].get_supportsname(), "Resource") self.assertEqual(scd.get_interfaces().get_interface()[0].get_name(), "Resource") self.assertEqual(scd.get_interfaces().get_interface()[0].get_inheritsinterface()[0].get_repid(), "IDL:CF/LifeCycle:1.0") # Verify that we can write the output and still be DTD valid tmpfile = tempfile.mktemp() try: tmp = open(tmpfile, "w") scd.export(tmp, 0) tmp.close() status = self._xmllint(tmpfile, "SCD") self.assertEqual(status, 0, "Python parser did not emit DTD compliant XML") finally: try: os.remove(tmpfile) except OSError: pass
def test_SADParser_usesdeviceref(self): sad = parsers.SADParser.parse("sdr/parser_tests/usesdeviceref.sad.xml") self.assertEqual(sad.get_id(), "colloc_usesdev_1") self.assertEqual(sad.get_name(), "colloc_usesdev") self.assertEqual(len(sad.componentfiles.get_componentfile()), 1) self.assertEqual(len(sad.partitioning.get_hostcollocation()), 1) colloc=sad.partitioning.get_hostcollocation()[0] self.assertEqual(len(colloc.get_componentplacement()),1) comp_place =colloc.get_componentplacement()[0] self.assertEqual(len(comp_place.get_componentinstantiation()),1) comp_ci=comp_place.get_componentinstantiation()[0] self.assertEqual(comp_ci.id_, "P1_1") self.assertEqual(comp_ci.get_usagename(), "P1_1") self.assertEqual(len(colloc.get_usesdeviceref()),1) udev_ref =colloc.get_usesdeviceref()[0] self.assertEqual(udev_ref.refid, "FrontEndTuner_1") # 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
def test_SADParser_devicerequires(self): sad = parsers.SADParser.parse("sdr/parser_tests/devicerequires.sad.xml") self.assertEqual(sad.get_id(), "device_requires_multicolor") self.assertEqual(sad.get_name(), "device_requires_multicolor") self.assertEqual(len(sad.componentfiles.get_componentfile()), 1) self.assertEqual(len(sad.partitioning.get_componentplacement()), 2) comp_place=sad.partitioning.get_componentplacement()[0] comp_in=comp_place.get_componentinstantiation()[0] self.assertEqual(comp_place.componentfileref.refid, "SimpleComponent_SPD_1") self.assertEqual(comp_in.id_, "SimpleComponent_Red") self.assertEqual(comp_in.get_usagename(), "SimpleComponent_Red") self.assertEqual(len(comp_in.devicerequires.get_requires()),2) self.assertEqual(comp_in.devicerequires.get_requires()[0].id, "color") self.assertEqual(comp_in.devicerequires.get_requires()[0].value, "RED") self.assertEqual(comp_in.devicerequires.get_requires()[1].id, "rank") self.assertEqual(comp_in.devicerequires.get_requires()[1].value, "15") comp_place=sad.partitioning.get_componentplacement()[1] comp_in=comp_place.get_componentinstantiation()[0] self.assertEqual(comp_place.componentfileref.refid, "SimpleComponent_SPD_1") self.assertEqual(comp_in.id_, "SimpleComponent_Green") self.assertEqual(comp_in.get_usagename(), "SimpleComponent_Green") self.assertEqual(len(comp_in.devicerequires.get_requires()),1) self.assertEqual(comp_in.devicerequires.get_requires()[0].id, "color") self.assertEqual(comp_in.devicerequires.get_requires()[0].value, "GREEN") # 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
def test_SADParser_loggingconfig(self): sad = parsers.SADParser.parse("sdr/parser_tests/loggingconfig.sad.xml") self.assertEqual(sad.get_id(), "device_requires_multicolor") self.assertEqual(sad.get_name(), "device_requires_multicolor") self.assertEqual(len(sad.componentfiles.get_componentfile()), 1) self.assertEqual(len(sad.partitioning.get_componentplacement()), 2) comp_place=sad.partitioning.get_componentplacement()[0] comp_in=comp_place.get_componentinstantiation()[0] self.assertEqual(comp_place.componentfileref.refid, "SimpleComponent_SPD_1") self.assertEqual(comp_in.id_, "SimpleComponent_Red") self.assertEqual(comp_in.get_usagename(), "SimpleComponent_Red") self.assertEqual(comp_in.loggingconfig.level, "ERROR") self.assertEqual(comp_in.loggingconfig.value, "path/to/my/log/file") comp_place=sad.partitioning.get_componentplacement()[1] comp_in=comp_place.get_componentinstantiation()[0] self.assertEqual(comp_place.componentfileref.refid, "SimpleComponent_SPD_1") self.assertEqual(comp_in.id_, "SimpleComponent_Green") self.assertEqual(comp_in.get_usagename(), "SimpleComponent_Green") self.assertEqual(comp_in.loggingconfig.value, "path/to/my/log/file2") # 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
def test_SADParser_affinityconfig(self): sad = parsers.SADParser.parse("sdr/parser_tests/affinity.sad.xml") self.assertEqual(sad.get_id(), "device_requires_multicolor") self.assertEqual(sad.get_name(), "device_requires_multicolor") self.assertEqual(len(sad.componentfiles.get_componentfile()), 1) self.assertEqual(len(sad.partitioning.get_componentplacement()), 1) comp_place=sad.partitioning.get_componentplacement()[0] comp_in=comp_place.get_componentinstantiation()[0] self.assertEqual(comp_place.componentfileref.refid, "SimpleComponent_SPD_1") self.assertEqual(comp_in.id_, "SimpleComponent_Red") self.assertEqual(comp_in.get_usagename(), "SimpleComponent_Red") self.assertEqual(comp_in.loggingconfig.level, "ERROR") self.assertEqual(comp_in.loggingconfig.value, "path/to/my/log/file") self.assertEqual(len(comp_in.affinity.get_simpleref()),2) self.assertEqual(comp_in.affinity.get_simpleref()[0].refid, "affinity::exec_directive_class") self.assertEqual(comp_in.affinity.get_simpleref()[0].value, "socket") self.assertEqual(comp_in.affinity.get_simpleref()[1].refid, "affinity::exec_directive_value") self.assertEqual(comp_in.affinity.get_simpleref()[1].value, "0") # 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
def test_DCDParser_loggingconfig(self): dcd = parsers.DCDParser.parse("sdr/parser_tests/loggingconfig.dcd.xml") self.assertEqual(dcd.get_id(), "test_GPP_green") self.assertEqual(dcd.get_name(), "test_GPP_green") self.assertEqual(len(dcd.componentfiles.get_componentfile()), 1) self.assertEqual(len(dcd.partitioning.get_componentplacement()), 1) gpp=dcd.partitioning.get_componentplacement()[0] gpp_ci=gpp.get_componentinstantiation()[0] self.assertEqual(gpp.get_componentfileref().get_refid(), "GPP1_file_1") self.assertEqual(gpp_ci.get_id(), "test_GPP_green::GPP_1") self.assertEqual(gpp_ci.get_usagename(), "test_GPP_green::GPP_1") self.assertEqual(gpp_ci.loggingconfig.level, "ERROR") self.assertEqual(gpp_ci.loggingconfig.value, "path/to/my/log/file") # Verify that we can write the output and still be DTD valid tmpfile = tempfile.mktemp() try: tmp = open(tmpfile, "w") dcd.export(tmp, 0) tmp.close() status = self._xmllint(tmpfile, "DCD") self.assertEqual(status, 0, "Python parser did not emit DTD compliant XML") finally: try: os.remove(tmpfile) except OSError: pass
def test_DCDParser_affinity(self): dcd = parsers.DCDParser.parse("sdr/parser_tests/affinity.dcd.xml") self.assertEqual(dcd.get_id(), "affinity_parse_1") self.assertEqual(dcd.get_name(), "test_affinity_node_socket") self.assertEqual(len(dcd.componentfiles.get_componentfile()), 1) self.assertEqual(len(dcd.partitioning.get_componentplacement()), 1) gpp=dcd.partitioning.get_componentplacement()[0] gpp_ci=gpp.get_componentinstantiation()[0] self.assertEqual(gpp.get_componentfileref().get_refid(), "GPP_File_1") self.assertEqual(gpp_ci.get_id(), "test_affinity_node:GPP_1") self.assertEqual(gpp_ci.get_usagename(), "GPP_1") self.assertEqual(len(gpp_ci.affinity.get_simpleref()),2) self.assertEqual(gpp_ci.affinity.get_simpleref()[0].refid, "affinity::exec_directive_class") self.assertEqual(gpp_ci.affinity.get_simpleref()[0].value, "socket") self.assertEqual(gpp_ci.affinity.get_simpleref()[1].refid, "affinity::exec_directive_value") self.assertEqual(gpp_ci.affinity.get_simpleref()[1].value, "0") # Verify that we can write the output and still be DTD valid tmpfile = tempfile.mktemp() try: tmp = open(tmpfile, "w") dcd.export(tmp, 0) tmp.close() status = self._xmllint(tmpfile, "DCD") self.assertEqual(status, 0, "Python parser did not emit DTD compliant XML") finally: try: os.remove(tmpfile) except OSError: pass
def setUp(self): self._dbfile = tempfile.mktemp() self._domainBooter_1, self._domainManager_1 = self.launchDomainManager(endpoint="giop:tcp::5679", dbURI=self._dbfile) self._domainBooter_2, self._domainManager_2 = launchDomain(2, self._root)
def load_logging_config_uri(orb, uri, binding=None): scheme, netloc, path, params, query, fragment = urlparse.urlparse(uri) if scheme == "file": ossie.utils.log4py.config.fileConfig(path, binding) elif scheme == "sca": q = dict([x.split("=") for x in query.split("&")]) try: fileSys = orb.string_to_object(q["fs"]) except KeyError: logging.warning("sca URI missing fs query parameter") else: if fileSys == None: logging.warning("Failed to lookup file system") else: try: t = tempfile.mktemp() tf = open(t, "w+") scaFile = fileSys.open(path, True) fileSize = scaFile.sizeOf() buf = scaFile.read(fileSize) tf.write(buf) tf.close() scaFile.close() ossie.utils.log4py.config.fileConfig(t) finally: os.remove(t) else: # Invalid scheme logging.warning("Invalid logging config URI scheme")
