kozorizki/gaphor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

More use of session instead of application

Browse Source
This commit is contained in:
Arjan Molenaar 2020-01-26 20:14:21 +01:00
parent 7417ed1d72
commit 2637281ec3
10 changed files with 174 additions and 274 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
examples/list_classes.py
View File

@ -8,7 +8,7 @@ import optparse
import sys
import gaphor.UML as UML
from gaphor.application import Application
from gaphor.application import Session
# Setup command line options.
usage = "usage: %prog [options] file.gaphor"
@ -33,7 +33,7 @@ if len(args) != 1:
model = args[0]
# Create the Gaphor application object.
session = Application.new_session()
session = Session()
# Get services we need.
element_factory = session.get_service("element_factory")

1
gaphor/UML/tests/test_properties.py
View File

@ -1,6 +1,5 @@
import pytest
from gaphor.application import Application
from gaphor.core import event_handler
from gaphor.UML.element import Element
from gaphor.UML.event import AssociationUpdated

19
gaphor/application.py
View File

@ -68,6 +68,9 @@ class _Application(Service, ActionProvider):
assert not self._services_by_name
uninitialized_services = load_services("gaphor.appservices", appservices)
self._services_by_name = init_services(uninitialized_services, application=self)
transaction.subscribers.add(self._transaction_proxy)
return self
def new_session(self, services=None):
@ -77,6 +80,7 @@ class _Application(Service, ActionProvider):
session = Session()
self.sessions.add(session)
self.active_session = session
return session
def has_sessions(self):
@ -95,6 +99,8 @@ class _Application(Service, ActionProvider):
This is mainly for testing purposes.
"""
transaction.subscribers.discard(self._transaction_proxy)
while self.sessions:
self.shutdown_session(self.sessions.pop())
@ -115,12 +121,17 @@ class _Application(Service, ActionProvider):
if self.active_session == session:
logger.info("Window not closed, abort quit operation")
return
self.shutdown()
def all(self, base: Type[T]) -> Iterator[Tuple[str, T]]:
return (
(n, c) for n, c in self._services_by_name.items() if isinstance(c, base)
)
def _transaction_proxy(self, event):
if self.active_session:
self.active_session.event_manager.handle(event)
class Session:
"""
@ -143,12 +154,6 @@ class Session:
self.component_registry.register(name, srv)
self.event_manager.handle(ServiceInitializedEvent(name, srv))
transaction.subscribers.add(self._transaction_proxy)
def _transaction_proxy(self, event):
if self is Application.active_session:
self.event_manager.handle(event)
def get_service(self, name):
if not self.component_registry:
raise NotInitializedError("Session is no longer alive")
@ -156,8 +161,6 @@ class Session:
return self.component_registry.get_service(name)
def shutdown(self):
transaction.subscribers.discard(self._transaction_proxy)
if self.component_registry:
for name, _srv in self.component_registry.all(Service):
self.shutdown_service(name)

1
gaphor/services/session.py
View File

@ -1,5 +1,4 @@
from gaphor.abc import Service
from gaphor.application import Application
class Session(Service):

5
gaphor/transaction.txt
View File

@ -8,7 +8,8 @@ Transaction support is located in module gaphor.transaction:
Do some basic initialization, so event emission will work:
>>> session = Application.new_session(services=['event_manager'])
>>> application = Application()
>>> session = application.new_session(services=['event_manager'])
>>> event_manager = session.get_service('event_manager')
The Transaction class is used mainly to signal the begin and end of a transaction. This is done by the TransactionBegin, TransactionCommit and TransactionRollback events:
@ -115,4 +116,4 @@ All transactions are marked for rollback once an exception is raised:
Cleanup:
>>> Application.shutdown()
>>> application.shutdown()

2
gaphor/ui/__init__.py
View File

@ -10,7 +10,7 @@ from typing import Optional
import gi
from gaphor.application import Application, Session, _Application
from gaphor.application import Application, _Application
from gaphor.core import event_handler
from gaphor.event import ActiveSessionChanged, SessionShutdown
from gaphor.ui.actiongroup import apply_application_actions

116
gaphor/ui/tests/test_diagrampage.py
View File

@ -1,63 +1,75 @@
import unittest
import pytest
from gaphas.examples import Box
from gaphor import UML
from gaphor.application import Application
from gaphor.application import Session
from gaphor.diagram.general.comment import CommentItem
from gaphor.ui.mainwindow import DiagramPage
class DiagramPageTestCase(unittest.TestCase):
def setUp(self):
session = Application.new_session(
services=[
"event_manager",
"component_registry",
"element_factory",
"main_window",
"properties",
"namespace",
"diagrams",
"toolbox",
"elementeditor",
"export_menu",
"tools_menu",
]
)
main_window = session.get_service("main_window")
main_window.open()
self.element_factory = session.get_service("element_factory")
self.diagram = self.element_factory.create(UML.Diagram)
self.page = DiagramPage(
self.diagram,
session.get_service("event_manager"),
self.element_factory,
session.get_service("properties"),
)
self.page.construct()
assert self.page.diagram == self.diagram
assert self.page.view.canvas == self.diagram.canvas
assert len(self.element_factory.lselect()) == 1
@pytest.fixture
def session():
session = Session(
services=[
"event_manager",
"component_registry",
"element_factory",
"main_window",
"properties",
"namespace",
"diagrams",
"toolbox",
"elementeditor",
"export_menu",
"tools_menu",
]
)
yield session
session.shutdown()
def tearDown(self):
self.page.close()
del self.page
self.diagram.unlink()
del self.diagram
Application.shutdown()
assert len(self.element_factory.lselect()) == 0
def test_creation(self):
pass
@pytest.fixture
def main_window(session):
main_window = session.get_service("main_window")
main_window.open()
def test_placement(self):
box = Box()
self.diagram.canvas.add(box)
self.diagram.canvas.update_now()
self.page.view.request_update([box])
self.diagram.create(
CommentItem, subject=self.element_factory.create(UML.Comment)
)
assert len(self.element_factory.lselect()) == 2
@pytest.fixture
def element_factory(session):
return session.get_service("element_factory")
@pytest.fixture
def diagram(element_factory):
diagram = element_factory.create(UML.Diagram)
yield diagram
diagram.unlink()
@pytest.fixture
def page(session, diagram, element_factory):
page = DiagramPage(
diagram,
session.get_service("event_manager"),
element_factory,
session.get_service("properties"),
)
page.construct()
assert page.diagram == diagram
assert page.view.canvas == diagram.canvas
yield page
page.close()
def test_creation(page, element_factory):
assert len(element_factory.lselect()) == 1
def test_placement(diagram, page, element_factory):
box = Box()
diagram.canvas.add(box)
diagram.canvas.update_now()
page.view.request_update([box])
diagram.create(CommentItem, subject=element_factory.create(UML.Comment))
assert len(element_factory.lselect()) == 2

288
gaphor/ui/tests/test_handletool.py
View File

@ -9,7 +9,7 @@ from gaphas.aspect import ConnectionSink, Connector
from gi.repository import Gdk, Gtk
from gaphor import UML
from gaphor.application import Application
from gaphor.application import Session
from gaphor.diagram.connectors import IConnect
from gaphor.diagram.diagramtools import ConnectHandleTool, DiagramItemConnector
from gaphor.diagram.general.comment import CommentItem
@ -21,7 +21,7 @@ from gaphor.ui.event import DiagramOpened
@pytest.fixture
def session():
session = Application.new_session(
session = Session(
services=[
"event_manager",
"component_registry",
@ -53,6 +53,12 @@ def event_manager(session):
return session.get_service("event_manager")
@pytest.fixture
def main_window(session):
main_window = session.get_service("main_window")
yield main_window
@pytest.fixture
def diagram(element_factory):
return element_factory.create(UML.Diagram)
@ -95,232 +101,112 @@ def test_connect(diagram, comment, commentline):
assert cinfo, cinfo
class HandleToolTestCase(unittest.TestCase):
def current_diagram_view(session):
"""
Handle connection tool integration tests.
Get a view for the current diagram.
"""
component_registry = session.get_service("component_registry")
view = component_registry.get(UIComponent, "diagrams").get_current_view()
def setUp(self):
self.session = Application.new_session(
services=[
"event_manager",
"component_registry",
"element_factory",
"main_window",
"properties_manager",
"properties",
"namespace",
"diagrams",
"toolbox",
"elementeditor",
"export_menu",
"tools_menu",
]
)
self.component_registry = self.session.get_service("component_registry")
self.event_manager = self.session.get_service("event_manager")
# realize view, forces bounding box recalculation
while Gtk.events_pending():
Gtk.main_iteration()
self.main_window = self.session.get_service("main_window")
self.main_window.open()
return view
def shutDown(self):
Application.shutdown()
def get_diagram_view(self, diagram):
"""
Get a view for diagram.
"""
view = self.component_registry.get(UIComponent, "diagrams").get_current_view()
def test_iconnect(session, event_manager, element_factory):
"""
Test basic glue functionality using CommentItem and CommentLine
items.
"""
diagram = element_factory.create(UML.Diagram)
event_manager.handle(DiagramOpened(diagram))
comment = diagram.create(CommentItem, subject=element_factory.create(UML.Comment))
# realize view, forces bounding box recalculation
while Gtk.events_pending():
Gtk.main_iteration()
actor = diagram.create(ActorItem, subject=element_factory.create(UML.Actor))
actor.matrix.translate(200, 200)
diagram.canvas.update_matrix(actor)
return view
line = diagram.create(CommentLineItem)
def test_iconnect(self):
"""
Test basic glue functionality using CommentItem and CommentLine
items.
"""
element_factory = self.session.get_service("element_factory")
diagram = element_factory.create(UML.Diagram)
self.event_manager.handle(DiagramOpened(diagram))
comment = diagram.create(
CommentItem, subject=element_factory.create(UML.Comment)
)
view = current_diagram_view(session)
assert view, "View should be available here"
comment_bb = view.get_item_bounding_box(comment)
actor = diagram.create(ActorItem, subject=element_factory.create(UML.Actor))
actor.matrix.translate(200, 200)
diagram.canvas.update_matrix(actor)
# select handle:
handle = line.handles()[-1]
tool = ConnectHandleTool(view=view)
line = diagram.create(CommentLineItem)
tool.grab_handle(line, handle)
handle.pos = (comment_bb.x, comment_bb.y)
item = tool.glue(line, handle, handle.pos)
assert item is not None
view = self.get_diagram_view(diagram)
assert view, "View should be available here"
comment_bb = view.get_item_bounding_box(comment)
tool.connect(line, handle, handle.pos)
cinfo = diagram.canvas.get_connection(handle)
assert cinfo.constraint is not None
assert cinfo.connected is actor, cinfo.connected
# select handle:
handle = line.handles()[-1]
tool = ConnectHandleTool(view=view)
Connector(line, handle).disconnect()
tool.grab_handle(line, handle)
handle.pos = (comment_bb.x, comment_bb.y)
item = tool.glue(line, handle, handle.pos)
assert item is not None
cinfo = diagram.canvas.get_connection(handle)
tool.connect(line, handle, handle.pos)
cinfo = diagram.canvas.get_connection(handle)
assert cinfo.constraint is not None
assert cinfo.connected is actor, cinfo.connected
assert cinfo is None
Connector(line, handle).disconnect()
cinfo = diagram.canvas.get_connection(handle)
def test_connect_comment_and_actor(session, event_manager, element_factory):
"""Test connect/disconnect on comment and actor using comment-line.
"""
diagram = element_factory.create(UML.Diagram)
event_manager.handle(DiagramOpened(diagram))
comment = diagram.create(CommentItem, subject=element_factory.create(UML.Comment))
assert cinfo is None
line = diagram.create(CommentLineItem)
def test_connect_comment_and_actor(self):
"""Test connect/disconnect on comment and actor using comment-line.
"""
element_factory = self.session.get_service("element_factory")
diagram = element_factory.create(UML.Diagram)
self.event_manager.handle(DiagramOpened(diagram))
comment = diagram.create(
CommentItem, subject=element_factory.create(UML.Comment)
)
view = current_diagram_view(session)
assert view, "View should be available here"
line = diagram.create(CommentLineItem)
tool = ConnectHandleTool(view)
view = self.get_diagram_view(diagram)
assert view, "View should be available here"
# Connect one end to the Comment:
handle = line.handles()[0]
tool.grab_handle(line, handle)
tool = ConnectHandleTool(view)
handle.pos = (0, 0)
sink = tool.glue(line, handle, handle.pos)
assert sink is not None
assert sink.item is comment
# Connect one end to the Comment:
handle = line.handles()[0]
tool.grab_handle(line, handle)
tool.connect(line, handle, handle.pos)
cinfo = diagram.canvas.get_connection(handle)
assert cinfo is not None, None
assert cinfo.item is line
assert cinfo.connected is comment
handle.pos = (0, 0)
sink = tool.glue(line, handle, handle.pos)
assert sink is not None
assert sink.item is comment
# Connect the other end to the Actor:
actor = diagram.create(ActorItem, subject=element_factory.create(UML.Actor))
tool.connect(line, handle, handle.pos)
cinfo = diagram.canvas.get_connection(handle)
assert cinfo is not None, None
assert cinfo.item is line
assert cinfo.connected is comment
handle = line.handles()[-1]
tool.grab_handle(line, handle)
# Connect the other end to the Actor:
actor = diagram.create(ActorItem, subject=element_factory.create(UML.Actor))
handle.pos = (0, 0)
sink = tool.glue(line, handle, handle.pos)
assert sink, f"No sink at {handle.pos}"
assert sink.item is actor
tool.connect(line, handle, handle.pos)
handle = line.handles()[-1]
tool.grab_handle(line, handle)
cinfo = view.canvas.get_connection(handle)
assert cinfo.item is line
assert cinfo.connected is actor
handle.pos = (0, 0)
sink = tool.glue(line, handle, handle.pos)
assert sink, f"No sink at {handle.pos}"
assert sink.item is actor
tool.connect(line, handle, handle.pos)
# Try to connect far away from any item will only do a full disconnect
assert len(comment.subject.annotatedElement) == 1, comment.subject.annotatedElement
assert actor.subject in comment.subject.annotatedElement
cinfo = view.canvas.get_connection(handle)
assert cinfo.item is line
assert cinfo.connected is actor
sink = tool.glue(line, handle, (500, 500))
assert sink is None, sink
tool.connect(line, handle, (500, 500))
# Try to connect far away from any item will only do a full disconnect
self.assertEqual(
len(comment.subject.annotatedElement), 1, comment.subject.annotatedElement
)
assert actor.subject in comment.subject.annotatedElement
sink = tool.glue(line, handle, (500, 500))
assert sink is None, sink
tool.connect(line, handle, (500, 500))
cinfo = view.canvas.get_connection(handle)
assert cinfo is None
def skiptest_connect_3(self):
"""Test connecting through events (button press/release, motion).
"""
element_factory = self.session.get_service("element_factory")
diagram = element_factory.create(UML.Diagram)
comment = diagram.create(
CommentItem, subject=element_factory.create(UML.Comment)
)
# self.assertEqual(30, comment.height)
# self.assertEqual(100, comment.width)
actor = diagram.create(ActorItem, subject=element_factory.create(UML.Actor))
actor.matrix.translate(200, 200)
diagram.canvas.update_matrix(actor)
# assert actor.height == 60, actor.height
# assert actor.width == 38, actor.width
line = diagram.create(CommentLineItem)
assert line.handles()[0].pos, (0.0, 0.0)
assert line.handles()[-1].pos, (10.0, 10.0)
view = self.get_diagram_view(diagram)
assert view, "View should be available here"
tool = ConnectHandleTool(view)
tool.on_button_press(Gdk.Event(x=0, y=0, state=0))
tool.on_button_release(Gdk.Event(x=0, y=0, state=0))
handle = line.handles()[0]
assert (0.0, 0.0) == view.canvas.get_matrix_i2c(line).transform_point(
*handle.pos
)
cinfo = diagram.canvas.get_connection(handle)
assert cinfo.connected is comment
# self.assertTrue(handle.connected_to is comment, 'c = ' + str(handle.connected_to))
# self.assertTrue(handle.connection_data is not None)
# Grab the second handle and drag it to the actor
tool.on_button_press(Gdk.Event(x=10, y=10, state=0))
tool.on_motion_notify(Gdk.Event(x=200, y=200, state=0xFFFF))
tool.on_button_release(Gdk.Event(x=200, y=200, state=0))
handle = line.handles()[-1]
assert (200, 200) == view.canvas.get_matrix_i2c(line).transform_point(
handle.x, handle.y
)
cinfo = diagram.canvas.get_connection(handle)
assert cinfo.connected is actor
# self.assertTrue(handle.connection_data is not None)
self.assertTrue(actor.subject in comment.subject.annotatedElement)
# Press, release, nothing should change
tool.on_button_press(Gdk.Event(x=200, y=200, state=0))
tool.on_motion_notify(Gdk.Event(x=200, y=200, state=0xFFFF))
tool.on_button_release(Gdk.Event(x=200, y=200, state=0))
handle = line.handles()[-1]
assert (200, 200) == view.canvas.get_matrix_i2c(line).transform_point(
handle.x, handle.y
)
cinfo = diagram.canvas.get_connection(handle)
assert cinfo.connected is actor
# self.assertTrue(handle.connection_data is not None)
self.assertTrue(actor.subject in comment.subject.annotatedElement)
# Move second handle away from the actor. Should remove connection
tool.on_button_press(Gdk.Event(x=200, y=200, state=0))
tool.on_motion_notify(Gdk.Event(x=500, y=500, state=0xFFFF))
tool.on_button_release(Gdk.Event(x=500, y=500, state=0))
handle = line.handles()[-1]
assert (500, 500) == view.canvas.get_matrix_i2c(line).transform_point(
handle.x, handle.y
)
cinfo = diagram.canvas.get_connection(handle)
assert cinfo is None
# self.assertTrue(handle.connection_data is None)
self.assertEqual(len(comment.subject.annotatedElement), 0)
cinfo = view.canvas.get_connection(handle)
assert cinfo is None

5
gaphor/ui/tests/test_lifecycle.py
View File

@ -15,8 +15,9 @@ class GtkApplicationStub:
@pytest.fixture
def application():
yield Application
Application.shutdown()
application = Application()
yield application
application.shutdown()
def two_sessions(application, gtk_app=GtkApplicationStub()):

7
tests/test_issue_53.py
View File

@ -1,16 +1,15 @@
import pytest
from gaphor import UML
from gaphor.application import Application, distribution
from gaphor.application import Session, distribution
from gaphor.storage.storage import load
@pytest.fixture
def session():
application = Application()
session = application.new_session()
session = Session()
yield session
application.shutdown()
session.shutdown()
@pytest.fixture