|
Packit |
6f3914 |
# -*- coding: utf-8 -*-
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
# Copyright (C) 2012-2018 Red Hat, Inc.
|
|
Packit |
6f3914 |
#
|
|
Packit |
6f3914 |
# This copyrighted material is made available to anyone wishing to use,
|
|
Packit |
6f3914 |
# modify, copy, or redistribute it subject to the terms and conditions of
|
|
Packit |
6f3914 |
# the GNU General Public License v.2, or (at your option) any later version.
|
|
Packit |
6f3914 |
# This program is distributed in the hope that it will be useful, but WITHOUT
|
|
Packit |
6f3914 |
# ANY WARRANTY expressed or implied, including the implied warranties of
|
|
Packit |
6f3914 |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General
|
|
Packit |
6f3914 |
# Public License for more details. You should have received a copy of the
|
|
Packit |
6f3914 |
# GNU General Public License along with this program; if not, write to the
|
|
Packit |
6f3914 |
# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
|
|
Packit |
6f3914 |
# 02110-1301, USA. Any Red Hat trademarks that are incorporated in the
|
|
Packit |
6f3914 |
# source code or documentation are not subject to the GNU General Public
|
|
Packit |
6f3914 |
# License and may only be used or replicated with the express permission of
|
|
Packit |
6f3914 |
# Red Hat, Inc.
|
|
Packit |
6f3914 |
#
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
"""Unit test dnf.lock module.
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
Locking is very hard to cover reasonably with a unit test, this is more or less
|
|
Packit |
6f3914 |
just a sanity check.
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
"""
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
from __future__ import absolute_import
|
|
Packit |
6f3914 |
from __future__ import unicode_literals
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
import multiprocessing
|
|
Packit |
6f3914 |
import os
|
|
Packit |
6f3914 |
import re
|
|
Packit |
6f3914 |
import threading
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
import dnf.lock
|
|
Packit |
6f3914 |
import dnf.pycomp
|
|
Packit |
6f3914 |
import dnf.util
|
|
Packit |
6f3914 |
from dnf.exceptions import ProcessLockError, ThreadLockError
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
import tests.support
|
|
Packit |
6f3914 |
from tests.support import mock
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
class ConcurrencyMixin(object):
|
|
Packit |
6f3914 |
def __init__(self, lock):
|
|
Packit |
6f3914 |
self.lock = lock
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
def run(self):
|
|
Packit |
6f3914 |
try:
|
|
Packit |
6f3914 |
with self.lock:
|
|
Packit |
6f3914 |
pass
|
|
Packit |
6f3914 |
except (ProcessLockError, ThreadLockError) as e:
|
|
Packit |
6f3914 |
self.queue.put(e)
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
class OtherThread(ConcurrencyMixin, threading.Thread):
|
|
Packit |
6f3914 |
def __init__(self, lock):
|
|
Packit |
6f3914 |
ConcurrencyMixin.__init__(self, lock)
|
|
Packit |
6f3914 |
threading.Thread.__init__(self)
|
|
Packit |
6f3914 |
self.queue = dnf.pycomp.Queue(1)
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
class OtherProcess(ConcurrencyMixin, multiprocessing.Process):
|
|
Packit |
6f3914 |
def __init__(self, lock):
|
|
Packit |
6f3914 |
ConcurrencyMixin.__init__(self, lock)
|
|
Packit |
6f3914 |
multiprocessing.Process.__init__(self)
|
|
Packit |
6f3914 |
self.queue = multiprocessing.Queue(1)
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
TARGET = os.path.join(tests.support.USER_RUNDIR, 'unit-test.pid')
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
def build_lock(blocking=False):
|
|
Packit |
6f3914 |
return dnf.lock.ProcessLock(TARGET, 'unit-tests', blocking)
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
class LockTest(tests.support.TestCase):
|
|
Packit |
6f3914 |
def test_fit_lock_dir(self):
|
|
Packit |
6f3914 |
orig = '/some'
|
|
Packit |
6f3914 |
with mock.patch('dnf.util.am_i_root', return_value=True):
|
|
Packit |
6f3914 |
self.assertEqual(dnf.lock._fit_lock_dir(orig), '/some')
|
|
Packit |
6f3914 |
with mock.patch('dnf.util.am_i_root', return_value=False):
|
|
Packit |
6f3914 |
dir_ = dnf.lock._fit_lock_dir(orig)
|
|
Packit |
6f3914 |
match = re.match(
|
|
Packit |
6f3914 |
r'/var/tmp/dnf-[^:]+-[^/]+/locks/2690814ff0269c86e5109d2915ea572092918cb3',
|
|
Packit |
6f3914 |
dir_)
|
|
Packit |
6f3914 |
self.assertTrue(match)
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
class ProcessLockTest(tests.support.TestCase):
|
|
Packit |
6f3914 |
@classmethod
|
|
Packit |
6f3914 |
def tearDownClass(cls):
|
|
Packit |
6f3914 |
dnf.util.rm_rf(tests.support.USER_RUNDIR)
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
def test_simple(self):
|
|
Packit |
6f3914 |
l1 = build_lock()
|
|
Packit |
6f3914 |
target = l1.target
|
|
Packit |
6f3914 |
with l1:
|
|
Packit |
6f3914 |
self.assertFile(target)
|
|
Packit |
6f3914 |
self.assertPathDoesNotExist(target)
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
def test_reentrance(self):
|
|
Packit |
6f3914 |
l1 = build_lock()
|
|
Packit |
6f3914 |
with l1:
|
|
Packit |
6f3914 |
with l1:
|
|
Packit |
6f3914 |
pass
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
def test_another_process(self):
|
|
Packit |
6f3914 |
l1 = build_lock()
|
|
Packit |
6f3914 |
process = OtherProcess(l1)
|
|
Packit |
6f3914 |
with l1:
|
|
Packit |
6f3914 |
process.start()
|
|
Packit |
6f3914 |
process.join()
|
|
Packit |
6f3914 |
self.assertIsInstance(process.queue.get(), ProcessLockError)
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
def test_another_process_blocking(self):
|
|
Packit |
6f3914 |
l1 = build_lock(blocking=True)
|
|
Packit |
6f3914 |
l2 = build_lock(blocking=True)
|
|
Packit |
6f3914 |
process = OtherProcess(l1)
|
|
Packit |
6f3914 |
target = l1.target
|
|
Packit |
6f3914 |
with l2:
|
|
Packit |
6f3914 |
process.start()
|
|
Packit |
6f3914 |
process.join()
|
|
Packit |
6f3914 |
self.assertEqual(process.queue.empty(), True)
|
|
Packit |
6f3914 |
self.assertPathDoesNotExist(target)
|
|
Packit |
6f3914 |
|
|
Packit |
6f3914 |
def test_another_thread(self):
|
|
Packit |
6f3914 |
l1 = build_lock()
|
|
Packit |
6f3914 |
thread = OtherThread(l1)
|
|
Packit |
6f3914 |
with l1:
|
|
Packit |
6f3914 |
thread.start()
|
|
Packit |
6f3914 |
thread.join()
|
|
Packit |
6f3914 |
self.assertIsInstance(thread.queue.get(), ThreadLockError)
|