Skip to content

Commit

Permalink
Another attempt - use copies of original test
Browse files Browse the repository at this point in the history
  • Loading branch information
@AlexanderWells-diamond
AlexanderWells-diamond committed Dec 2, 2024
1 parent e3f0e00 commit 14d7300
Showing 1 changed file with 217 additions and 1 deletion.
218 changes: 217 additions & 1 deletion tests/test_autosave.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@



def test_context_manager_thread_safety(tmp_path):
def create_many_threads(tmp_path):
autosave.configure(tmp_path, DEVICE_NAME)
in_cm_event = threading.Event()

Expand All @@ -31,3 +31,219 @@ def create_pv_in_thread(name):
in_cm_event.set()
builder.aOut("PV-FROM-CM")
[x.join() for x in threads]

def original_test_1(tmp_path):
autosave.configure(tmp_path, DEVICE_NAME)
in_cm_event = threading.Event()

def create_pv_in_thread(name):
in_cm_event.wait()
builder.aOut(name, autosave=False)
pv_thread_before_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-BEFORE"])
pv_thread_in_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-DURING"])
pv_thread_before_cm.start()
with autosave.Autosave(["VAL", "EGU"]):
in_cm_event.set()
builder.aOut("PV-FROM-CM")
pv_thread_in_cm.start()
pv_thread_in_cm.join()
pv_thread_before_cm.join()

assert "PV-FROM-THREAD-BEFORE" not in autosave.Autosave._pvs
assert "PV-FROM-THREAD-DURING" not in autosave.Autosave._pvs
assert device_core.LookupRecord("PV-FROM-THREAD-BEFORE")
assert device_core.LookupRecord("PV-FROM-THREAD-DURING")

def original_test_2(tmp_path):
autosave.configure(tmp_path, DEVICE_NAME)
in_cm_event = threading.Event()

def create_pv_in_thread(name):
in_cm_event.wait()
builder.aOut(name, autosave=False)
pv_thread_before_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-BEFORE"])
pv_thread_in_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-DURING"])
pv_thread_before_cm.start()
with autosave.Autosave(["VAL", "EGU"]):
in_cm_event.set()
builder.aOut("PV-FROM-CM")
pv_thread_in_cm.start()
pv_thread_in_cm.join()
pv_thread_before_cm.join()

assert "PV-FROM-THREAD-BEFORE" not in autosave.Autosave._pvs
assert "PV-FROM-THREAD-DURING" not in autosave.Autosave._pvs
assert device_core.LookupRecord("PV-FROM-THREAD-BEFORE")
assert device_core.LookupRecord("PV-FROM-THREAD-DURING")

def original_test_3(tmp_path):
autosave.configure(tmp_path, DEVICE_NAME)
in_cm_event = threading.Event()

def create_pv_in_thread(name):
in_cm_event.wait()
builder.aOut(name, autosave=False)
pv_thread_before_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-BEFORE"])
pv_thread_in_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-DURING"])
pv_thread_before_cm.start()
with autosave.Autosave(["VAL", "EGU"]):
in_cm_event.set()
builder.aOut("PV-FROM-CM")
pv_thread_in_cm.start()
pv_thread_in_cm.join()
pv_thread_before_cm.join()

assert "PV-FROM-THREAD-BEFORE" not in autosave.Autosave._pvs
assert "PV-FROM-THREAD-DURING" not in autosave.Autosave._pvs
assert device_core.LookupRecord("PV-FROM-THREAD-BEFORE")
assert device_core.LookupRecord("PV-FROM-THREAD-DURING")

def original_test_4(tmp_path):
autosave.configure(tmp_path, DEVICE_NAME)
in_cm_event = threading.Event()

def create_pv_in_thread(name):
in_cm_event.wait()
builder.aOut(name, autosave=False)
pv_thread_before_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-BEFORE"])
pv_thread_in_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-DURING"])
pv_thread_before_cm.start()
with autosave.Autosave(["VAL", "EGU"]):
in_cm_event.set()
builder.aOut("PV-FROM-CM")
pv_thread_in_cm.start()
pv_thread_in_cm.join()
pv_thread_before_cm.join()

assert "PV-FROM-THREAD-BEFORE" not in autosave.Autosave._pvs
assert "PV-FROM-THREAD-DURING" not in autosave.Autosave._pvs
assert device_core.LookupRecord("PV-FROM-THREAD-BEFORE")
assert device_core.LookupRecord("PV-FROM-THREAD-DURING")

def original_test_5(tmp_path):
autosave.configure(tmp_path, DEVICE_NAME)
in_cm_event = threading.Event()

def create_pv_in_thread(name):
in_cm_event.wait()
builder.aOut(name, autosave=False)
pv_thread_before_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-BEFORE"])
pv_thread_in_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-DURING"])
pv_thread_before_cm.start()
with autosave.Autosave(["VAL", "EGU"]):
in_cm_event.set()
builder.aOut("PV-FROM-CM")
pv_thread_in_cm.start()
pv_thread_in_cm.join()
pv_thread_before_cm.join()

assert "PV-FROM-THREAD-BEFORE" not in autosave.Autosave._pvs
assert "PV-FROM-THREAD-DURING" not in autosave.Autosave._pvs
assert device_core.LookupRecord("PV-FROM-THREAD-BEFORE")
assert device_core.LookupRecord("PV-FROM-THREAD-DURING")

def original_test_6(tmp_path):
autosave.configure(tmp_path, DEVICE_NAME)
in_cm_event = threading.Event()

def create_pv_in_thread(name):
in_cm_event.wait()
builder.aOut(name, autosave=False)
pv_thread_before_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-BEFORE"])
pv_thread_in_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-DURING"])
pv_thread_before_cm.start()
with autosave.Autosave(["VAL", "EGU"]):
in_cm_event.set()
builder.aOut("PV-FROM-CM")
pv_thread_in_cm.start()
pv_thread_in_cm.join()
pv_thread_before_cm.join()

assert "PV-FROM-THREAD-BEFORE" not in autosave.Autosave._pvs
assert "PV-FROM-THREAD-DURING" not in autosave.Autosave._pvs
assert device_core.LookupRecord("PV-FROM-THREAD-BEFORE")
assert device_core.LookupRecord("PV-FROM-THREAD-DURING")

def original_test_7(tmp_path):
autosave.configure(tmp_path, DEVICE_NAME)
in_cm_event = threading.Event()

def create_pv_in_thread(name):
in_cm_event.wait()
builder.aOut(name, autosave=False)
pv_thread_before_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-BEFORE"])
pv_thread_in_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-DURING"])
pv_thread_before_cm.start()
with autosave.Autosave(["VAL", "EGU"]):
in_cm_event.set()
builder.aOut("PV-FROM-CM")
pv_thread_in_cm.start()
pv_thread_in_cm.join()
pv_thread_before_cm.join()

assert "PV-FROM-THREAD-BEFORE" not in autosave.Autosave._pvs
assert "PV-FROM-THREAD-DURING" not in autosave.Autosave._pvs
assert device_core.LookupRecord("PV-FROM-THREAD-BEFORE")
assert device_core.LookupRecord("PV-FROM-THREAD-DURING")

def original_test_8(tmp_path):
autosave.configure(tmp_path, DEVICE_NAME)
in_cm_event = threading.Event()

def create_pv_in_thread(name):
in_cm_event.wait()
builder.aOut(name, autosave=False)
pv_thread_before_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-BEFORE"])
pv_thread_in_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-DURING"])
pv_thread_before_cm.start()
with autosave.Autosave(["VAL", "EGU"]):
in_cm_event.set()
builder.aOut("PV-FROM-CM")
pv_thread_in_cm.start()
pv_thread_in_cm.join()
pv_thread_before_cm.join()

assert "PV-FROM-THREAD-BEFORE" not in autosave.Autosave._pvs
assert "PV-FROM-THREAD-DURING" not in autosave.Autosave._pvs
assert device_core.LookupRecord("PV-FROM-THREAD-BEFORE")
assert device_core.LookupRecord("PV-FROM-THREAD-DURING")

def original_test_9(tmp_path):
autosave.configure(tmp_path, DEVICE_NAME)
in_cm_event = threading.Event()

def create_pv_in_thread(name):
in_cm_event.wait()
builder.aOut(name, autosave=False)
pv_thread_before_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-BEFORE"])
pv_thread_in_cm = threading.Thread(
target=create_pv_in_thread, args=["PV-FROM-THREAD-DURING"])
pv_thread_before_cm.start()
with autosave.Autosave(["VAL", "EGU"]):
in_cm_event.set()
builder.aOut("PV-FROM-CM")
pv_thread_in_cm.start()
pv_thread_in_cm.join()
pv_thread_before_cm.join()

assert "PV-FROM-THREAD-BEFORE" not in autosave.Autosave._pvs
assert "PV-FROM-THREAD-DURING" not in autosave.Autosave._pvs
assert device_core.LookupRecord("PV-FROM-THREAD-BEFORE")
assert device_core.LookupRecord("PV-FROM-THREAD-DURING")

0 comments on commit 14d7300

Please sign in to comment.