feat(booking): wip booking test

Author: vmscw

tests/test_api.py

@@ -185,7 +185,7 @@ def test_create_and_cancel_booking():
     try:
         now = datetime.now(timezone.utc)
         booking_start = now + timedelta(seconds=15)
-        booking_finish = booking_start + timedelta(hours=1)
+        booking_finish = booking_start + timedelta(seconds=20)
         booking_description = "my lovely booking"
 
         session = client.create_session(
@@ -231,7 +231,7 @@ def test_create_and_cancel_booking():
 
             print(booking.status)
 
-            assert booking.status in ["cancelling", "cancelled"]
+            assert booking.status in ["validated", "cancelling", "cancelled"]
 
             if booking.status == "cancelled":
                 break
@@ -240,18 +240,18 @@ def test_create_and_cancel_booking():
 def test_create_overlaping_booking():
     client = _get_client()
 
-    try:
-        platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
+    platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
 
-        assert len(platforms) > 0
+    assert len(platforms) > 0
 
-        target_platform = platforms[0]
+    target_platform = platforms[0]
 
-        now = datetime.now(timezone.utc)
-        booking_start = now + timedelta(days=1)
-        booking_finish = booking_start + timedelta(hours=1)
-        booking_description = "my overlaping booking"
+    now = datetime.now(timezone.utc)
+    booking_start = now + timedelta(days=1)
+    booking_finish = booking_start + timedelta(hours=1)
+    booking_description = "my overlaping booking"
 
+    try:
         session = client.create_session(
             platform_id=target_platform.id,
             booking_demand_started_at=booking_start,
@@ -260,143 +260,237 @@ def test_create_overlaping_booking():
         )
 
         assert session == None
-    finally:
-        client.delete_session(session.id)
+    except Exception as e:
+        assert "existing booking" in str(e)
 
 
 def test_create_too_long_booking():
     client = _get_client()
 
-    try:
-        platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
+    platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
 
-        assert len(platforms) > 0
+    assert len(platforms) > 0
 
-        target_platform = platforms[0]
+    target_platform = platforms[0]
 
-        now = datetime.now(timezone.utc)
-        booking_start = now + timedelta(minutes=1)
-        booking_finish = booking_start + timedelta(hours=10)
-        booking_description = "my too long booking"
+    now = datetime.now(timezone.utc)
+    booking_start = now + timedelta(minutes=1)
+    booking_finish = booking_start + timedelta(hours=10)
+    booking_description = "my too long booking"
 
-        session = client.create_session(
+    try:
+        _ = client.create_session(
             platform_id=target_platform.id,
             booking_demand_started_at=booking_start,
             booking_demand_finished_at=booking_finish,
             booking_demand_description=booking_description,
         )
-
-        assert session == None
-    finally:
-        client.delete_session(session.id)
+    except Exception as e:
+        assert "max_duration not respected" in str(e)
 
 
 def test_create_too_short_booking():
     client = _get_client()
 
-    try:
-        platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
+    platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
 
-        assert len(platforms) > 0
+    assert len(platforms) > 0
 
-        target_platform = platforms[0]
+    target_platform = platforms[0]
 
-        now = datetime.now(timezone.utc)
-        booking_start = now + timedelta(minutes=1)
-        booking_finish = booking_start + timedelta(seconds=10)
-        booking_description = "my lovely booking"
+    now = datetime.now(timezone.utc)
+    booking_start = now + timedelta(minutes=1)
+    booking_finish = booking_start + timedelta(seconds=10)
+    booking_description = "my lovely booking"
 
-        session = client.create_session(
+    try:
+        _ = client.create_session(
             platform_id=target_platform.id,
             booking_demand_started_at=booking_start,
             booking_demand_finished_at=booking_finish,
             booking_demand_description=booking_description,
         )
-
-        assert session == None
-    finally:
-        client.delete_session(session.id)
+    except Exception as e:
+        assert "min_duration not respected" in str(e)
 
 
-def test_create_too_short_booking():
+def test_create_too_far_booking():
     client = _get_client()
 
-    try:
-        platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
+    platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
 
-        assert len(platforms) > 0
+    assert len(platforms) > 0
 
-        target_platform = platforms[0]
+    target_platform = platforms[0]
 
-        now = datetime.now(timezone.utc)
-        booking_start = now + timedelta(minutes=1)
-        booking_finish = booking_start + timedelta(seconds=10)
-        booking_description = "my lovely booking"
+    now = datetime.now(timezone.utc)
+    booking_start = now + timedelta(days=30)
+    booking_finish = booking_start + timedelta(hours=2)
+    booking_description = "my lovely booking"
 
-        session = client.create_session(
+    try:
+        _ = client.create_session(
             platform_id=target_platform.id,
             booking_demand_started_at=booking_start,
             booking_demand_finished_at=booking_finish,
             booking_demand_description=booking_description,
         )
+    except Exception as e:
+        assert "max_planification_duration not respected" in str(e)
 
-        assert session == None
-    finally:
-        client.delete_session(session.id)
 
-
-def test_create_too_far_booking():
+def test_create_too_close_booking():
     client = _get_client()
 
-    try:
-        platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
+    platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
 
-        assert len(platforms) > 0
+    assert len(platforms) > 0
 
-        target_platform = platforms[0]
+    target_platform = platforms[0]
 
-        now = datetime.now(timezone.utc)
-        booking_start = now + timedelta(days=30)
-        booking_finish = booking_start + timedelta(hours=2)
-        booking_description = "my lovely booking"
+    now = datetime.now(timezone.utc)
+    booking_start = now + timedelta(seconds=10)
+    booking_finish = booking_start + timedelta(hours=2)
+    booking_description = "my lovely booking"
 
-        session = client.create_session(
+    try:
+        _ = client.create_session(
             platform_id=target_platform.id,
             booking_demand_started_at=booking_start,
             booking_demand_finished_at=booking_finish,
             booking_demand_description=booking_description,
         )
+    except Exception as e:
+        assert "min_planification_duration not respected" in str(e)
 
-        assert session == None
-    finally:
-        client.delete_session(session.id)
 
+def test_cannot_start_not_booked_session_during_booked_session_is_running():
+    client = _get_client()
 
-def test_create_too_close_booking():
+    platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
+
+    assert len(platforms) > 0
+
+    target_platform = platforms[0]
+
+    now = datetime.now(timezone.utc)
+    booking_start = now + timedelta(seconds=20)
+    booking_finish = booking_start + timedelta(seconds=20)
+    booking_description = "my booking"
+
+    booked_session = client.create_session(
+        platform_id=target_platform.id,
+        booking_demand_started_at=booking_start,
+        booking_demand_finished_at=booking_finish,
+        booking_demand_description=booking_description,
+    )
+
+    time.sleep(23)
+
+    not_booked_session = client.create_session(
+        platform_id=target_platform.id,
+    )
+
+    while True:
+        time.sleep(2)
+        not_booked_session = client.get_session(session_id=not_booked_session.id)
+
+        booked_session = client.get_session(session_id=booked_session.id)
+
+        assert not_booked_session.status in ["starting"]
+        assert booked_session.status in ["starting", "running", "stopping", "stopped"]
+
+        if booked_session.status == "stopped":
+            break
+
+    while True:
+        time.sleep(2)
+        not_booked_session = client.get_session(session_id=not_booked_session.id)
+
+        assert not_booked_session.status in ["starting", "running"]
+
+        if booked_session.status == "running":
+            break
+
+    client.delete_session(not_booked_session.id)
+
+
+def test_not_booked_session_is_killed_when_booked_session_starts():
     client = _get_client()
 
+    platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
+
+    assert len(platforms) > 0
+
+    target_platform = platforms[0]
+
     try:
-        platforms = client.list_platforms(name=_TEST_PLATFORM_NAME)
+        not_booked_session_1 = client.create_session(
+            platform_id=target_platform.id,
+        )
 
-        assert len(platforms) > 0
+        assert not_booked_session_1
+
+        not_booked_session_2 = client.create_session(
+            platform_id=target_platform.id,
+        )
 
-        target_platform = platforms[0]
+        assert not_booked_session_2
 
         now = datetime.now(timezone.utc)
-        booking_start = now + timedelta(seconds=10)
-        booking_finish = booking_start + timedelta(hours=2)
-        booking_description = "my lovely booking"
+        booking_start = now + timedelta(seconds=20)
+        booking_finish = booking_start + timedelta(seconds=20)
+        booking_description = "my booking"
 
-        session = client.create_session(
+        booked_session = client.create_session(
             platform_id=target_platform.id,
             booking_demand_started_at=booking_start,
             booking_demand_finished_at=booking_finish,
             booking_demand_description=booking_description,
         )
 
-        assert session == None
+        while True:
+            time.sleep(2)
+            not_booked_session_1 = client.get_session(
+                session_id=not_booked_session_1.id
+            )
+            not_booked_session_2 = client.get_session(
+                session_id=not_booked_session_2.id
+            )
+
+            assert not_booked_session_1.status in ["starting", "running"]
+            assert not_booked_session_2.status in ["starting", "running"]
+
+            if (
+                not_booked_session_1.status == "running"
+                and not_booked_session_2.status == "running"
+            ):
+                break
+
+        time.sleep(25)
+
+        while True:
+            time.sleep(2)
+            not_booked_session_1 = client.get_session(
+                session_id=not_booked_session_1.id
+            )
+            not_booked_session_2 = client.get_session(
+                session_id=not_booked_session_2.id
+            )
+
+            assert not_booked_session_1.status in ["running", "stopping", "stopped"]
+            assert not_booked_session_2.status in ["running", "stopping", "stopped"]
+
+            if (
+                not_booked_session_1.status == "stopped"
+                and not_booked_session_2.status == "stopped"
+            ):
+                break
+
     finally:
-        client.delete_session(session.id)
+        client.delete_session(not_booked_session_1.id)
+        client.delete_session(not_booked_session_2.id)
+        client.delete_session(booked_session.id)
 
 
 # def test_create_session_same_deduplication_id():