Use SesionAlreadyOpen/NoSessionOpen.

ognibuild/session/__init__.py

@@ -21,6 +21,20 @@ import sys
 import subprocess
 
 
+class NoSessionOpen(Exception):
+    """There is no session open."""
+
+    def __init__(self, session):
+        self.session = session
+
+
+class SessionAlreadyOpen(Exception):
+    """There is already a session open."""
+
+    def __init__(self, session):
+        self.session = session
+
+
 class Session(object):
     def __enter__(self) -> "Session":
         return self

ognibuild/session/plain.py

@@ -16,7 +16,7 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
-from . import Session
+from . import Session, NoSessionOpen, SessionAlreadyOpen
 
 import contextlib
 import os
@@ -30,7 +30,12 @@ class PlainSession(Session):
 
     location = "/"
 
+    def __init__(self):
+        self.es = None
+
     def _prepend_user(self, user, args):
+        if self.es is None:
+            raise NoSessionOpen(self)
         if user is not None:
             import getpass
             if user != getpass.getuser():
@@ -41,12 +46,17 @@ class PlainSession(Session):
         return "%s()" % (type(self).__name__, )
 
     def __enter__(self) -> "Session":
+        if self.es is not None:
+            raise SessionAlreadyOpen(self)
         self.es = contextlib.ExitStack()
         self.es.__enter__()
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):
+        if self.es is None:
+            raise NoSessionOpen(self)
         self.es.__exit__(exc_type, exc_val, exc_tb)
+        self.es = None
         return False
 
     def create_home(self):

ognibuild/session/schroot.py

@@ -24,7 +24,7 @@ import tempfile
 from typing import Optional, List, Dict
 
 
-from . import Session, SessionSetupFailure
+from . import Session, SessionSetupFailure, NoSessionOpen, SessionAlreadyOpen
 
 
 class SchrootSession(Session):
@@ -39,8 +39,11 @@ class SchrootSession(Session):
         self.chroot = chroot
         self._location = None
         self._cwd = None
+        self.session_id = None
 
     def _get_location(self) -> str:
+        if self.session_id is None:
+            raise NoSessionOpen(self)
         return (
             subprocess.check_output(
                 ["schroot", "--location", "-c", "session:" + self.session_id]
@@ -50,9 +53,14 @@ class SchrootSession(Session):
         )
 
     def _end_session(self) -> None:
+        if self.session_id is None:
+            raise NoSessionOpen(self)
         subprocess.check_output(["schroot", "-c", "session:" + self.session_id, "-e"])
+        self.session_id = None
 
     def __enter__(self) -> "Session":
+        if self.es is not None:
+            raise SessionAlreadyOpen(self)
         try:
             self.session_id = (
                 subprocess.check_output(["schroot", "-c", self.chroot, "-b"])
@@ -87,6 +95,8 @@ class SchrootSession(Session):
         user: Optional[str] = None,
         env: Optional[Dict[str, str]] = None,
     ):
+        if self.session_id is None:
+            raise NoSessionOpen(self)
         base_argv = ["schroot", "-r", "-c", "session:" + self.session_id]
         if cwd is None:
             cwd = self._cwd