本文整理汇总了Python中synapse.cortex.openurl函数的典型用法代码示例。如果您正苦于以下问题：Python openurl函数的具体用法？Python openurl怎么用？Python openurl使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了openurl函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_cortex_sync
def test_cortex_sync(self):
core0 = s_cortex.openurl('ram://')
core1 = s_cortex.openurl('ram://')
core0.on('core:sync', core1.sync )
tufo0 = core0.formTufoByProp('foo','bar',baz='faz')
tufo1 = core1.getTufoByProp('foo','bar')
self.assertEqual( tufo1[1].get('foo'), 'bar' )
self.assertEqual( tufo1[1].get('foo:baz'), 'faz' )
tufo0 = core0.addTufoTag(tufo0,'hehe')
tufo1 = core1.getTufoByProp('foo','bar')
self.assertTrue( s_tags.tufoHasTag(tufo1,'hehe') )
core0.delTufoTag(tufo0,'hehe')
tufo1 = core1.getTufoByProp('foo','bar')
self.assertFalse( s_tags.tufoHasTag(tufo1,'hehe') )
core0.delTufo(tufo0)
tufo1 = core1.getTufoByProp('foo','bar')
self.assertIsNone( tufo1 )
开发者ID:pombredanne，项目名称:synapse-3，代码行数:26，代码来源:test_cortex.py
示例2: test_pki_base
def test_pki_base(self):
core0 = s_cortex.openurl('ram:///')
pki0 = s_pki.PkiStor(core0)
uidn = guid()
root = pki0.genRootToken(bits=512, save=True)
dork = pki0.genUserToken('[email protected]', bits=512, save=True)
visi = pki0.genUserToken('[email protected]', can=('sign:cert',), bits=512, save=True)
iden = root[0]
self.assertIsNotNone( iden )
self.assertEqual(visi[1].get('user'), '[email protected]')
vcrt = pki0.genTokenCert(visi, signas=iden)
pki0.setTokenCert(visi[0], vcrt, save=True)
dcrt = s_pki.initTokenCert(dork)
dcrt = pki0.signTokenCert(visi[0], dcrt)
core1 = s_cortex.openurl('ram:///')
pki1 = s_pki.PkiStor(core1)
self.assertIsNone( pki1.loadCertToken( dcrt ) )
pki1.setTokenTufo(root)
self.assertIsNotNone( pki1.loadCertToken( dcrt ) )
foob = b'foob'
iden = root[0]
sign = pki0.genByteSign(iden, b'blob')
toks = tuple( pki0.iterTokenTufos() )
self.assertEqual( len(toks), 3)
self.assertTrue( pki1.isValidSign(iden, sign, b'blob') )
self.assertFalse( pki1.isValidSign(iden, sign, foob) )
self.assertFalse( pki1.isValidSign(foob, sign, b'blob') )
self.assertFalse( pki1.isValidSign(iden, foob, b'blob') )
pki0.fini()
core0.fini()
pki1.fini()
core1.fini()
开发者ID:imjonsnooow，项目名称:synapse，代码行数:50，代码来源:test_pki.py
示例3: test_cortex_ramhost
def test_cortex_ramhost(self):
core0 = s_cortex.openurl('ram:///foobar')
core1 = s_cortex.openurl('ram:///foobar')
self.assertEqual( id(core0), id(core1) )
core0.fini()
core0 = s_cortex.openurl('ram:///foobar')
core1 = s_cortex.openurl('ram:///bazfaz')
self.assertNotEqual( id(core0), id(core1) )
core0.fini()
core1.fini()
core0 = s_cortex.openurl('ram:///')
core1 = s_cortex.openurl('ram:///')
self.assertNotEqual( id(core0), id(core1) )
core0.fini()
core1.fini()
core0 = s_cortex.openurl('ram://')
core1 = s_cortex.openurl('ram://')
self.assertNotEqual( id(core0), id(core1) )
core0.fini()
core1.fini()
开发者ID:pombredanne，项目名称:synapse-3，代码行数:30，代码来源:test_cortex.py
示例4: benchmark_cortex
def benchmark_cortex(test_data, url, cleanup_func, num_threads=1):
core = s_cortex.openurl(url)
_prepopulate_core(core, test_data.prepop_rows)
g = {'_addRows': _addRows, '_getTufosByIdens': _getTufosByIdens, 'core': core,
'test_data': test_data, '_getTufoByPropVal': _getTufoByPropVal, '_run_x': _run_x}
do_it('_addRows', 'test_data.rows', num_threads, g, 1, 1, len(test_data.rows))
if cleanup_func:
del core
core = s_cortex.openurl(url)
g['core'] = core
do_it('_getTufosByIdens', 'test_data.idens', num_threads, g, 2, 5, NUM_TUFOS)
do_it('_getTufoByPropVal', 'test_data.props', num_threads, g, 2, 5, NUM_TUFOS)
if cleanup_func:
cleanup_func()
开发者ID:vivisect，项目名称:synapse，代码行数:15，代码来源:benchmark_cortex.py
示例5: test_datamodel_cortex
def test_datamodel_cortex(self):
core = s_cortex.openurl('ram:///')
core.addTufoForm('foo')
core.addTufoProp('foo', 'bar', ptype='int', defval=10)
core.formTufoByProp('foo','hehe')
core.formTufoByProp('foo','haha')
core.formTufoByProp('foo','blah', bar=99)
tufo0 = core.formTufoByProp('foo','hehe')
self.assertEqual( tufo0[1].get('foo:bar'), 10 )
core.setTufoProp(tufo0,'bar',30)
self.assertEqual( tufo0[1].get('foo:bar'), 30 )
tufo1 = core.formTufoByProp('foo','hehe')
self.assertEqual( tufo0[0], tufo1[0] )
tufos = core.getTufosByProp('foo')
self.assertEqual( len(tufos) , 3 )
tufos = core.getTufosByProp('foo:bar', valu=30, limit=20)
self.assertEqual( len(tufos) , 1 )
tufos = core.getTufosByProp('foo:bar', valu=99, limit=20)
self.assertEqual( len(tufos) , 1 )
开发者ID:k1derly-fe，项目名称:synapse，代码行数:29，代码来源:test_datamodel.py
示例6: test_cortex_caching_del_tufo
def test_cortex_caching_del_tufo(self):
with s_cortex.openurl('ram://') as core:
tufo0 = core.formTufoByProp('foo','bar', qwer=10)
tufo1 = core.formTufoByProp('foo','baz', qwer=10)
core.setConfOpt('caching',1)
tufs0 = core.getTufosByProp('foo:qwer')
tufs1 = core.getTufosByProp('foo:qwer', valu=10)
tufs2 = core.getTufosByProp('foo:qwer', valu=11)
self.assertEqual( len(tufs0), 2 )
self.assertEqual( len(tufs1), 2 )
self.assertEqual( len(tufs2), 0 )
core.delTufo( tufo0 )
#tufo2 = core.formTufoByProp('foo','lol', qwer=10)
tufs0 = core.getTufosByProp('foo:qwer')
tufs1 = core.getTufosByProp('foo:qwer', valu=10)
tufs2 = core.getTufosByProp('foo:qwer', valu=11)
self.assertEqual( len(tufs0), 1 )
self.assertEqual( len(tufs1), 1 )
self.assertEqual( len(tufs2), 0 )
开发者ID:williballenthin，项目名称:synapse，代码行数:27，代码来源:test_cortex.py
示例7: test_cortex_by_type
def test_cortex_by_type(self):
with s_cortex.openurl('ram://') as core:
core.addTufoForm('foo')
core.addTufoProp('foo','min', ptype='time:epoch:min')
core.addTufoProp('foo','max', ptype='time:epoch:max')
core.addTufoForm('bar')
core.addTufoProp('bar','min', ptype='time:epoch:min')
core.addTufoProp('bar','max', ptype='time:epoch:max')
core.addTufoForm('baz')
core.addTufoProp('baz','min', ptype='time:epoch')
core.addTufoProp('baz','max', ptype='time:epoch')
props = {'min':20,'max':20}
tufo0 = core.formTufoByProp('foo', 'hurr', **props)
tufo1 = core.formTufoByProp('bar', 'durr', **props)
tufo2 = core.formTufoByProp('baz', 'durr', **props)
want = tuple(sorted([tufo0[0],tufo1[0]]))
res0 = core.getTufosByPropType('time:epoch:min', valu=20)
self.eq( tuple(sorted([r[0] for r in res0])), want )
开发者ID:williballenthin，项目名称:synapse，代码行数:26，代码来源:test_cortex.py
示例8: test_sess_current
def test_sess_current(self):
core = s_cortex.openurl('ram:///')
cura = s_session.Curator(core=core)
sess = cura.new()
iden = sess.iden
sess.put('woot',10)
with sess:
woot = s_session.current()
self.eq(sess.iden,woot.iden)
self.eq( woot.get('woot'), 10 )
sess.put('haha',30,save=False)
self.eq( woot.get('haha'), 30 )
cura.fini()
cura = s_session.Curator(core=core)
sess = cura.get(iden)
self.eq(sess.get('woot'), 10)
self.eq(sess.get('haha'), None)
core.fini()
开发者ID:williballenthin，项目名称:synapse，代码行数:30，代码来源:test_session.py
示例9: test_cortex_stats
def test_cortex_stats(self):
rows = [
(guid(), 'foo:bar', 1, 99),
(guid(), 'foo:bar', 2, 99),
(guid(), 'foo:bar', 3, 99),
(guid(), 'foo:bar', 5, 99),
(guid(), 'foo:bar', 8, 99),
(guid(), 'foo:bar', 13, 99),
(guid(), 'foo:bar', 21, 99),
]
core = s_cortex.openurl('ram://')
core.addRows(rows)
self.assertEqual( core.getStatByProp('sum','foo:bar'), 53 )
self.assertEqual( core.getStatByProp('count','foo:bar'), 7 )
self.assertEqual( core.getStatByProp('min','foo:bar'), 1 )
self.assertEqual( core.getStatByProp('max','foo:bar'), 21 )
self.assertEqual( core.getStatByProp('average','foo:bar'), 7.571428571428571 )
self.assertEqual( core.getStatByProp('any','foo:bar'), True)
self.assertEqual( core.getStatByProp('all','foo:bar'), True)
histo = core.getStatByProp('histo','foo:bar')
self.assertEqual( histo.get(13), 1 )
开发者ID:pombredanne，项目名称:synapse-3，代码行数:27，代码来源:test_cortex.py
示例10: test_cortex_stats
def test_cortex_stats(self):
rows = [
(guid(), "foo:bar", 1, 99),
(guid(), "foo:bar", 2, 99),
(guid(), "foo:bar", 3, 99),
(guid(), "foo:bar", 5, 99),
(guid(), "foo:bar", 8, 99),
(guid(), "foo:bar", 13, 99),
(guid(), "foo:bar", 21, 99),
]
core = s_cortex.openurl("ram://")
core.addRows(rows)
self.assertEqual(core.getStatByProp("sum", "foo:bar"), 53)
self.assertEqual(core.getStatByProp("count", "foo:bar"), 7)
self.assertEqual(core.getStatByProp("min", "foo:bar"), 1)
self.assertEqual(core.getStatByProp("max", "foo:bar"), 21)
self.assertEqual(core.getStatByProp("average", "foo:bar"), 7.571428571428571)
self.assertEqual(core.getStatByProp("any", "foo:bar"), True)
self.assertEqual(core.getStatByProp("all", "foo:bar"), True)
histo = core.getStatByProp("histo", "foo:bar")
self.assertEqual(histo.get(13), 1)
开发者ID:imjonsnooow，项目名称:synapse，代码行数:27，代码来源:test_cortex.py
示例11: test_cortex_fire_set
def test_cortex_fire_set(self):
core = s_cortex.openurl('ram://')
props = {'foo:bar':'lol'}
tufo = core.formTufoByProp('foo', 'hehe', bar='lol')
events = ['tufo:set','tufo:props:foo','tufo:set:foo:bar']
wait = self.getTestWait(core,len(events),*events)
core.setTufoProps(tufo,bar='hah')
evts = wait.wait()
self.assertEqual( evts[0][0], 'tufo:set')
self.assertEqual( evts[0][1]['tufo'][0], tufo[0])
self.assertEqual( evts[0][1]['props']['foo:bar'], 'hah' )
self.assertEqual( evts[1][0], 'tufo:props:foo')
self.assertEqual( evts[1][1]['tufo'][0], tufo[0])
self.assertEqual( evts[1][1]['props']['foo:bar'], 'hah' )
self.assertEqual( evts[2][0], 'tufo:set:foo:bar')
self.assertEqual( evts[2][1]['tufo'][0], tufo[0])
self.assertEqual( evts[2][1]['valu'], 'hah' )
core.fini()
开发者ID:pombredanne，项目名称:synapse-3，代码行数:27，代码来源:test_cortex.py
示例12: test_cortex_caching_under_limit
def test_cortex_caching_under_limit(self):
with s_cortex.openurl('ram://') as core:
tufo0 = core.formTufoByProp('foo','bar', qwer=10)
tufo1 = core.formTufoByProp('foo','baz', qwer=10)
core.setConfOpt('caching',1)
tufs0 = core.getTufosByProp('foo:qwer', limit=9)
tufs1 = core.getTufosByProp('foo:qwer', valu=10, limit=9)
self.assertEqual( len(tufs0), 2 )
self.assertEqual( len(tufs1), 2 )
# when an entry is deleted from a cache result that was under it's limit
# it should be removed but *not* invalidated
core.delTufo(tufo0)
self.assertIsNotNone( core.cache_bykey.get( ('foo:qwer',None,9) ) )
self.assertIsNotNone( core.cache_bykey.get( ('foo:qwer',10,9) ) )
tufs0 = core.getTufosByProp('foo:qwer', limit=9)
tufs1 = core.getTufosByProp('foo:qwer', valu=10, limit=9)
self.assertEqual( len(tufs0), 1 )
self.assertEqual( len(tufs1), 1 )
开发者ID:williballenthin，项目名称:synapse，代码行数:28，代码来源:test_cortex.py
示例13: test_cortex_tags
def test_cortex_tags(self):
core = s_cortex.openurl('ram://')
core.addTufoForm('foo')
hehe = core.formTufoByProp('foo','hehe')
core.addTufoTag(hehe,'lulz.rofl')
lulz = core.getTufoByProp('syn:tag','lulz')
self.assertIsNone( lulz[1].get('syn:tag:up') )
self.assertEqual( lulz[1].get('syn:tag:doc'), '')
self.assertEqual( lulz[1].get('syn:tag:title'), '')
self.assertEqual( lulz[1].get('syn:tag:depth'), 0 )
rofl = core.getTufoByProp('syn:tag','lulz.rofl')
self.assertEqual( rofl[1].get('syn:tag:doc'), '')
self.assertEqual( rofl[1].get('syn:tag:title'), '')
self.assertEqual( rofl[1].get('syn:tag:up'), 'lulz' )
self.assertEqual( rofl[1].get('syn:tag:depth'), 1 )
core.delTufo(lulz)
# tag and subs should be wiped
self.assertIsNone( core.getTufoByProp('syn:tag','lulz') )
self.assertIsNone( core.getTufoByProp('syn:tag','lulz.rofl') )
self.assertEqual( len(core.getTufosByTag('foo','lulz')), 0 )
self.assertEqual( len(core.getTufosByTag('foo','lulz.rofl')), 0 )
core.fini()
开发者ID:k1derly-fe，项目名称:synapse，代码行数:34，代码来源:test_cortex.py
示例14: test_cortex_tufo_pop
def test_cortex_tufo_pop(self):
with s_cortex.openurl('ram://') as core:
foo0 = core.formTufoByProp('foo','bar',woot='faz')
foo1 = core.formTufoByProp('foo','baz',woot='faz')
self.assertEqual( 2, len(core.popTufosByProp('foo:woot', valu='faz')))
self.assertEqual( 0, len(core.getTufosByProp('foo')))
开发者ID:pombredanne，项目名称:synapse-3，代码行数:7，代码来源:test_cortex.py
示例15: test_cortex_caching
def test_cortex_caching(self):
with s_cortex.openurl('ram://') as core:
tufo0 = core.formTufoByProp('foo','bar', asdf=2)
tufo1 = core.formTufoByProp('foo','baz', asdf=2)
answ0 = core.getTufosByProp('foo')
answ1 = core.getTufosByProp('foo', valu='bar')
self.eq( len(answ0), 2 )
self.eq( len(answ1), 1 )
self.eq( len(core.cache_fifo), 0 )
self.eq( len(core.cache_bykey), 0 )
self.eq( len(core.cache_byiden), 0 )
self.eq( len(core.cache_byprop), 0 )
core.setConfOpt('caching',1)
self.eq( core.caching, 1 )
answ0 = core.getTufosByProp('foo')
self.eq( len(answ0), 2 )
self.eq( len(core.cache_fifo), 1 )
self.eq( len(core.cache_bykey), 1 )
self.eq( len(core.cache_byiden), 2 )
self.eq( len(core.cache_byprop), 1 )
开发者ID:williballenthin，项目名称:synapse，代码行数:29，代码来源:test_cortex.py
示例16: getSwarmEnv
def getSwarmEnv(self):
tenv = TestEnv()
core0 = s_cortex.openurl("ram://")
core1 = s_cortex.openurl("ram://")
tenv.add("core0", core0, fini=True)
tenv.add("core1", core1, fini=True)
tufo0 = core0.formTufoByProp("foo:bar", "baz", vvv="visi")
tufo1 = core0.formTufoByProp("foo:bar", "faz", vvv="visi")
tufo2 = core1.formTufoByProp("foo:bar", "lol", vvv="visi")
tufo3 = core1.formTufoByProp("foo:bar", "hai", vvv="visi")
tufo4 = core0.formTufoByProp("zzz:woot", 10, vvv="visi")
tufo5 = core1.formTufoByProp("zzz:woot", 12, vvv="romp")
tenv.add("tufo0", tufo0)
tenv.add("tufo1", tufo1)
tenv.add("tufo2", tufo2)
tenv.add("tufo3", tufo3)
dmon = s_daemon.Daemon()
link = dmon.listen("tcp://127.0.0.1:0")
tenv.add("link", link)
tenv.add("dmon", dmon, fini=True)
port = link[1].get("port")
svcbus = s_service.SvcBus()
tenv.add("svcbus", svcbus, fini=True)
dmon.share("syn.svcbus", svcbus)
svcrmi = s_telepath.openurl("tcp://127.0.0.1/syn.svcbus", port=port)
tenv.add("svcrmi", svcrmi, fini=True)
s_service.runSynSvc("cortex", core0, svcrmi, tags=("hehe.haha",))
s_service.runSynSvc("cortex", core1, svcrmi, tags=("hehe.hoho",))
runt = s_runtime.Runtime(svcrmi)
tenv.add("runt", runt, fini=True)
return tenv
开发者ID:williballenthin，项目名称:synapse，代码行数:46，代码来源:test_swarm_runtime.py
示例17: getSwarmEnv
def getSwarmEnv(self):
tenv = TestEnv()
core0 = s_cortex.openurl('ram://')
core1 = s_cortex.openurl('ram://')
tenv.add('core0',core0,fini=True)
tenv.add('core1',core1,fini=True)
tufo0 = core0.formTufoByProp('foo:bar','baz',vvv='visi')
tufo1 = core0.formTufoByProp('foo:bar','faz',vvv='visi')
tufo2 = core1.formTufoByProp('foo:bar','lol',vvv='visi')
tufo3 = core1.formTufoByProp('foo:bar','hai',vvv='visi')
tufo4 = core0.formTufoByProp('zzz:woot',10,vvv='visi')
tufo5 = core1.formTufoByProp('zzz:woot',12,vvv='romp')
tenv.add('tufo0',tufo0)
tenv.add('tufo1',tufo1)
tenv.add('tufo2',tufo2)
tenv.add('tufo3',tufo3)
dmon = s_daemon.Daemon()
link = dmon.listen('tcp://127.0.0.1:0')
tenv.add('link',link)
tenv.add('dmon',dmon,fini=True)
port = link[1].get('port')
svcbus = s_service.SvcBus()
tenv.add('svcbus',svcbus,fini=True)
dmon.share('syn.svcbus',svcbus)
svcrmi = s_telepath.openurl('tcp://127.0.0.1/syn.svcbus', port=port)
tenv.add('svcrmi',svcrmi,fini=True)
s_service.runSynSvc('cortex',core0,svcrmi,tags=('hehe.haha',))
s_service.runSynSvc('cortex',core1,svcrmi,tags=('hehe.hoho',))
runt = s_runtime.Runtime(svcrmi)
tenv.add('runt',runt,fini=True)
return tenv
开发者ID:jhsmith，项目名称:synapse，代码行数:46，代码来源:test_swarm_runtime.py
示例18: test_cortex_tufo_setprop
def test_cortex_tufo_setprop(self):
core = s_cortex.openurl('ram://')
foob = core.formTufoByProp('foo','bar',baz='faz')
self.assertEqual( foob[1].get('foo:baz'), 'faz' )
core.setTufoProp(foob,'baz','zap')
self.assertEqual( len(core.getTufosByProp('foo:baz',valu='zap')), 1 )
开发者ID:pombredanne，项目名称:synapse-3，代码行数:8，代码来源:test_cortex.py
示例19: test_cortex_tufo_setprop
def test_cortex_tufo_setprop(self):
core = s_cortex.openurl("ram://")
foob = core.formTufoByProp("foo", "bar", baz="faz")
self.assertEqual(foob[1].get("foo:baz"), "faz")
core.setTufoProp(foob, "baz", "zap")
self.assertEqual(len(core.getTufosByProp("foo:baz", valu="zap")), 1)
开发者ID:imjonsnooow，项目名称:synapse，代码行数:8，代码来源:test_cortex.py
示例20: test_cortex_dict
def test_cortex_dict(self):
core = s_cortex.openurl('ram://')
core.addTufoForm('foo:bar', ptype='int')
core.addTufoForm('baz:faz', ptype='int', defval=22)
modl = core.getModelDict()
self.assertEqual( modl['forms'][-2:], ['foo:bar','baz:faz'] )
self.assertEqual( modl['props']['foo:bar'][1]['ptype'], 'int')
self.assertEqual( modl['props']['baz:faz'][1]['defval'], 22)
开发者ID:k1derly-fe，项目名称:synapse，代码行数:9，代码来源:test_cortex.py
注：本文中的synapse.cortex.openurl函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论