Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"short": "node types",
"long": "set of node types to include in query result",
}
}
}
},
"cmdline": {
"en": "{\"nodes\":${nodes}, \"relationships\": ${relationships}}",
"script": "{\"nodes\":${nodes}, \"relationships\": ${relationships}}"
},
}"""
q1 = ClientQuery("allqueries", metadata1)
q1.validate_json()
q2 = ClientQuery("allqueries", metadata2)
q2.validate_json()
q3 = ClientQuery("ipowners", metadata3)
q3.validate_json()
q4 = ClientQuery("subgraph", metadata4)
q4.validate_json()
print("LOADING TREE!")
dirname = os.path.dirname(sys.argv[0])
dirname = "." if dirname == "" else dirname
queries = ClientQuery.load_tree(qstore, "%s/../queries" % dirname)
qstore.db_transaction = None
oops = ServiceUnavailable("OOPS?")
for _ in range(10):
try:
qstore.db_transaction = qstore.db.begin(autocommit=False)
break
except ServiceUnavailable as oops:
def _validate_nodetype(name, _paraminfo, value):
"""validate a node type - ignoring case"""
ret = ClientQuery._get_nodetype(value)
if ret is not None:
return ret
raise ValueError("Value of %s [%s] is not a known node type" % (name, value))
def _validate_hostname(name, paraminfo, value):
"""Validate a hostname value"""
return ClientQuery._validate_dnsname(name, paraminfo, value)
}
}
}
},
"cmdline": {
"en": "{\"nodes\":${nodes}, \"relationships\": ${relationships}}",
"script": "{\"nodes\":${nodes}, \"relationships\": ${relationships}}"
},
}"""
q1 = ClientQuery("allqueries", metadata1)
q1.validate_json()
q2 = ClientQuery("allqueries", metadata2)
q2.validate_json()
q3 = ClientQuery("ipowners", metadata3)
q3.validate_json()
q4 = ClientQuery("subgraph", metadata4)
q4.validate_json()
print("LOADING TREE!")
dirname = os.path.dirname(sys.argv[0])
dirname = "." if dirname == "" else dirname
queries = ClientQuery.load_tree(qstore, "%s/../queries" % dirname)
qstore.db_transaction = None
oops = ServiceUnavailable("OOPS?")
for _ in range(10):
try:
qstore.db_transaction = qstore.db.begin(autocommit=False)
break
except ServiceUnavailable as oops:
time.sleep(1)
if qstore.db_transaction is None:
# message W0212: access to protected member of client class
# pylint: disable=W0212
ClientQuery._validationmethods = {
"int": ClientQuery._validate_int,
"float": ClientQuery._validate_float,
"bool": ClientQuery._validate_bool,
"string": ClientQuery._validate_string,
"enum": ClientQuery._validate_enum,
"ipaddr": ClientQuery._validate_ipaddr,
"list": ClientQuery._validate_list,
"macaddr": ClientQuery._validate_macaddr,
"hostname": ClientQuery._validate_hostname,
"dnsname": ClientQuery._validate_dnsname,
"regex": ClientQuery._validate_regex,
"nodetype": ClientQuery._validate_nodetype,
"reltype": ClientQuery._validate_reltype,
}
if __name__ == "__main__":
# pylint: disable=C0413,C0411
import inject
from cmainit import CMAInjectables
inject.configure_once(CMAInjectables.test_config_injection)
@inject.params(qstore="Store")
def testcode(qstore=None):
"""
:param qstore: Store
:return: None
"""
def _validate_reltype(name, _paraminfo, value):
"""Validate a relationship type - ignoring case"""
ret = ClientQuery._get_reltype(value)
if ret is not None:
return ret
raise ValueError("Value of %s [%s] is not a known relationship type" % (name, value))
designation_s = str(designation)
relstr = reltype_expr(reltypes)
nodestr = str(nodetypes)
query = PythonDroneSubgraphQuery.basequery % (designation_s, relstr, nodestr, nodestr)
# print('RUNNING THIS QUERY:', query, file=stderr)
for row in self.store.load_cypher_query(query):
yield row
# message W0212: access to protected member of client class
# pylint: disable=W0212
ClientQuery._validationmethods = {
"int": ClientQuery._validate_int,
"float": ClientQuery._validate_float,
"bool": ClientQuery._validate_bool,
"string": ClientQuery._validate_string,
"enum": ClientQuery._validate_enum,
"ipaddr": ClientQuery._validate_ipaddr,
"list": ClientQuery._validate_list,
"macaddr": ClientQuery._validate_macaddr,
"hostname": ClientQuery._validate_hostname,
"dnsname": ClientQuery._validate_dnsname,
"regex": ClientQuery._validate_regex,
"nodetype": ClientQuery._validate_nodetype,
"reltype": ClientQuery._validate_reltype,
}
if __name__ == "__main__":
# pylint: disable=C0413,C0411
import inject
from cmainit import CMAInjectables
designation = [designation]
designation_s = str(designation)
relstr = reltype_expr(reltypes)
nodestr = str(nodetypes)
query = PythonDroneSubgraphQuery.basequery % (designation_s, relstr, nodestr, nodestr)
# print('RUNNING THIS QUERY:', query, file=stderr)
for row in self.store.load_cypher_query(query):
yield row
# message W0212: access to protected member of client class
# pylint: disable=W0212
ClientQuery._validationmethods = {
"int": ClientQuery._validate_int,
"float": ClientQuery._validate_float,
"bool": ClientQuery._validate_bool,
"string": ClientQuery._validate_string,
"enum": ClientQuery._validate_enum,
"ipaddr": ClientQuery._validate_ipaddr,
"list": ClientQuery._validate_list,
"macaddr": ClientQuery._validate_macaddr,
"hostname": ClientQuery._validate_hostname,
"dnsname": ClientQuery._validate_dnsname,
"regex": ClientQuery._validate_regex,
"nodetype": ClientQuery._validate_nodetype,
"reltype": ClientQuery._validate_reltype,
}
if __name__ == "__main__":
# pylint: disable=C0413,C0411
import inject
from cmainit import CMAInjectables
if isinstance(designation, (str, unicode)):
designation = [designation]
designation_s = str(designation)
relstr = reltype_expr(reltypes)
nodestr = str(nodetypes)
query = PythonDroneSubgraphQuery.basequery % (designation_s, relstr, nodestr, nodestr)
# print('RUNNING THIS QUERY:', query, file=stderr)
for row in self.store.load_cypher_query(query):
yield row
# message W0212: access to protected member of client class
# pylint: disable=W0212
ClientQuery._validationmethods = {
"int": ClientQuery._validate_int,
"float": ClientQuery._validate_float,
"bool": ClientQuery._validate_bool,
"string": ClientQuery._validate_string,
"enum": ClientQuery._validate_enum,
"ipaddr": ClientQuery._validate_ipaddr,
"list": ClientQuery._validate_list,
"macaddr": ClientQuery._validate_macaddr,
"hostname": ClientQuery._validate_hostname,
"dnsname": ClientQuery._validate_dnsname,
"regex": ClientQuery._validate_regex,
"nodetype": ClientQuery._validate_nodetype,
"reltype": ClientQuery._validate_reltype,
}
if __name__ == "__main__":
# pylint: disable=C0413,C0411
import inject
"type": "nodetype"
},
"lang": {
"en": {
"short": "node types",
"long": "set of node types to include in query result",
}
}
}
},
"cmdline": {
"en": "{\"nodes\":${nodes}, \"relationships\": ${relationships}}",
"script": "{\"nodes\":${nodes}, \"relationships\": ${relationships}}"
},
}"""
q1 = ClientQuery("allqueries", metadata1)
q1.validate_json()
q2 = ClientQuery("allqueries", metadata2)
q2.validate_json()
q3 = ClientQuery("ipowners", metadata3)
q3.validate_json()
q4 = ClientQuery("subgraph", metadata4)
q4.validate_json()
print("LOADING TREE!")
dirname = os.path.dirname(sys.argv[0])
dirname = "." if dirname == "" else dirname
queries = ClientQuery.load_tree(qstore, "%s/../queries" % dirname)
qstore.db_transaction = None
oops = ServiceUnavailable("OOPS?")
for _ in range(10):