From aba0be7960fb99620174743cf4de8f0e134410d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aur=C3=A9lien=20Delobelle?= Date: Sat, 5 Jan 2019 19:19:58 +0100 Subject: [PATCH] kfet.test -- Isolate kfet_open in testcases to avoid cache collisions --- kfet/open/tests.py | 35 ++++++++++++++++++++++------------- 1 file changed, 22 insertions(+), 13 deletions(-) diff --git a/kfet/open/tests.py b/kfet/open/tests.py index 75a9bf8a..b4481994 100644 --- a/kfet/open/tests.py +++ b/kfet/open/tests.py @@ -1,4 +1,5 @@ import json +import threading from datetime import timedelta from unittest import mock @@ -8,7 +9,7 @@ from django.contrib.auth.models import AnonymousUser, Permission, User from django.test import Client from django.utils import timezone -from . import OpenKfet, kfet_open +from . import OpenKfet from .consumers import OpenKfetConsumer @@ -16,10 +17,10 @@ class OpenKfetTest(ChannelTestCase): """OpenKfet object unit-tests suite.""" def setUp(self): - self.kfet_open = OpenKfet() - - def tearDown(self): - self.kfet_open.clear_cache() + self.kfet_open = OpenKfet( + cache_prefix="test_kfetopen_%s" % threading.get_ident() + ) + self.addCleanup(self.kfet_open.clear_cache) def test_defaults(self): """Default values.""" @@ -136,8 +137,14 @@ class OpenKfetViewsTest(ChannelTestCase): self.c_a = Client() self.c_a.login(username="admin", password="admin") - def tearDown(self): - kfet_open.clear_cache() + self.kfet_open = OpenKfet( + cache_prefix="test_kfetopen_%s" % threading.get_ident() + ) + self.addCleanup(self.kfet_open.clear_cache) + + views_patcher = mock.patch("kfet.open.views.kfet_open", self.kfet_open) + views_patcher.start() + self.addCleanup(views_patcher.stop) def test_door(self): """Edit raw_status.""" @@ -146,14 +153,14 @@ class OpenKfetViewsTest(ChannelTestCase): "/k-fet/open/raw_open", {"raw_open": sent, "token": "plop"} ) self.assertEqual(200, resp.status_code) - self.assertEqual(expected, kfet_open.raw_open) + self.assertEqual(expected, self.kfet_open.raw_open) def test_force_close(self): """Edit force_close.""" for sent, expected in [(1, True), (0, False)]: resp = self.c_a.post("/k-fet/open/force_close", {"force_close": sent}) self.assertEqual(200, resp.status_code) - self.assertEqual(expected, kfet_open.force_close) + self.assertEqual(expected, self.kfet_open.force_close) def test_force_close_forbidden(self): """Can't edit force_close without kfet.can_force_close permission.""" @@ -236,8 +243,10 @@ class OpenKfetScenarioTest(ChannelTestCase): self.r_c_ws = WSClient() self.r_c_ws.force_login(self.r) - def tearDown(self): - kfet_open.clear_cache() + self.kfet_open = OpenKfet( + cache_prefix="test_kfetopen_%s" % threading.get_ident() + ) + self.addCleanup(self.kfet_open.clear_cache) def ws_connect(self, ws_client): ws_client.send_and_consume( @@ -288,8 +297,8 @@ class OpenKfetScenarioTest(ChannelTestCase): def test_scenario_2(self): """Starting falsely closed, clients connect, disable force close.""" - kfet_open.raw_open = True - kfet_open.force_close = True + self.kfet_open.raw_open = True + self.kfet_open.force_close = True msg = self.ws_connect(self.c_ws) self.assertEqual(OpenKfet.CLOSED, msg["status"])