Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from py2neo.ogm import GraphObject, Label, Property, RelatedTo, RelatedFrom
class MovieGraphObject(GraphObject):
pass
class Film(MovieGraphObject):
__primarylabel__ = "Movie"
__primarykey__ = "title"
awesome = Label()
musical = Label()
science_fiction = Label(name="SciFi")
title = Property()
tag_line = Property(key="tagline", default="Bit boring")
year_of_release = Property(key="released")
actors = RelatedFrom("Person", "ACTED_IN")
assert thing.__node__.identity is not None
def test_simple_push(graph):
thing = SimpleThing()
graph.push(thing)
assert thing.__node__.graph is graph
assert thing.__node__.identity is not None
class A(GraphObject):
b = RelatedTo("B")
class B(GraphObject):
a = RelatedFrom("A")
def test_crossover(graph):
a = A()
b = B()
a.b.add(b)
b.a.add(a)
graph.create(a)
graph.create(b)
__primarykey__ = "uuid"
uuid = Property()
projname = Property()
qs = Property()
Parameter = RelatedTo("KeyValuePair")
def __init__(self, projname=None, qs=None):
self.projname=projname
self.qs = qs
self.uuid = str(uuid4())
class HeaderList(GraphObject):
__primarykey__ = "uuid"
uuid = Property()
projname = Property()
Header = RelatedTo("KeyValuePair")
def __init__(self, projname=None):
self.projname=projname
self.uuid = str(uuid4())
class Body(GraphObject):
url = Property()
Scheme = RelatedTo("DataValue")
Netloc = RelatedTo("DataValue")
Params = RelatedTo("DataValue")
Fragment = RelatedTo("DataValue")
Path = RelatedTo("DataValue")
QueryString = RelatedTo("KeyValuePair")
def __init__(self, projname=None, url=None):
self.projname = projname
self.url = url
self.uuid = str(uuid4())
class HeaderList(GraphObject):
__primarykey__ = "uuid"
uuid = Property()
projname = Property()
Header = RelatedTo("KeyValuePair")
def __init__(self, projname=None):
self.projname = projname
self.uuid = str(uuid4())
class Body(GraphObject):
value = RelatedTo("DataValue")
def __init__(self, content, projname=None):
self.uuid = "{}".format(datetime.datetime.now())
self.projname = projname
self.value.add(DataValue(value=content))
def getValue(self):
return list(self.value)[0].value
def inject(self, graph):
pass
class SessionElementBoolean(GraphObject, SessionElement):
type = SessionElementType.boolean
__primarykey__ = "uuid"
projname = Property()
uuid = Property()
value = RelatedTo("DataValue")
def __init__(self, content, projname=None):
self.uuid = "{}".format(datetime.datetime.now())
self.projname = projname
self.value.add(DataValue(value=content))
def getValue(self):
return list(self.value)[0].value
Contains = RelatedTo("DataValue")
def __init__(self, projname=None, ctype=None):
super(Body, self).__init__()
self.content_type = ctype
self.uuid = str(uuid4())
"""
Observation: Observations have a sequence number and a timestamp.
An observation is unique.
"""
class SeleneseCommand(GraphObject):
__primarykey__ = "uuid"
uuid = Property()
projname = Property()
session = Property()
user = Property()
seq = Property()
ts = Property()
command = Property()
target = Property()
value = Property()
Command = RelatedTo("DataValue")
Next = RelatedTo("SeleneseCommand")
Transaction = RelatedTo("HTTPResponse")
Caused = RelatedTo("SQLQuery")
def __init__(self, projname=None, session=None, user=None, seq=None, ts=None, method=None, url=None):
self.projname = projname
self.session = session
self.user = user
self.seq = seq
self.ts = ts
self.method = method
self.url = url
self.uuid = "{} [{} {}] {}.{}.{}".format(type(self).__name__, seq, ts, projname, session, user)
class HTTPResponse(GraphObject):
__primarykey__ = "uuid"
uuid = Property()
projname = Property()
session = Property()
user = Property()
seq = Property()
ts = Property()
status = Property()
Header = RelatedTo("HeaderList")
Body = RelatedTo("Body") # We can use any other type of node. Apparently this library does not to type enforcement for nodes.
projname = Property()
uuid = Property()
def __init__(self, projname=None):
self.uuid = "{}".format(datetime.datetime.now())
self.projname = projname
def getValue(self):
return None
def inject(self, graph):
graph.push(self)
class SessionElementString(GraphObject, SessionElement):
type = SessionElementType.string
__primarykey__ = "uuid"
uuid = Property()
projname = Property()
value = RelatedTo("DataValue")
def __init__(self, content, projname=None):
self.uuid = "{}".format(datetime.datetime.now())
self.projname = projname
self.value.add(DataValue(value=content))
def getValue(self):
return list(self.value)[0].value
class SessionElementType(Enum):
string = 1
integer = 2
array = 3
empty = 4
boolean = 5
class SessionElement():
type = None
content = None
class SessionElementEmpty(GraphObject, SessionElement):
type = SessionElementType.empty
__primarykey__ = "uuid"
projname = Property()
uuid = Property()
def __init__(self, projname=None):
self.uuid = "{}".format(datetime.datetime.now())
self.projname = projname
def getValue(self):
return None
def inject(self, graph):
graph.push(self)
class SQLTokenList(GraphObject):
__primarykey__ = "uuid"
uuid = Property()
Child = RelatedTo("SQLToken")
def __init__(self, projname=None):
self.projname = projname
self.uuid = str(uuid4())
class SQLStatement(GraphObject):
__primarykey__ = "uuid"
uuid = Property()
ttype = Property()
stmt = Property()
Child = RelatedTo("SQLToken")
def __init__(self, projname=None, statement=None):
self.projname = projname
self.stmt = statement
self.uuid = str(uuid4())