import struct

import sys

import unittest

from multiprocessing import Process

from test.support import (verbose, TESTFN, unlink, run_unittest, import_module,

cpython_only)

# Skip test if no fcntl module.

fcntl = import_module('fcntl')

# TODO - Write tests for flock() and lockf().

def get_lockdata():

try:

self.assertRaises(ValueError, fcntl.flock, -1, fcntl.LOCK_SH)

self.assertRaises(TypeError, fcntl.flock, 'spam', fcntl.LOCK_SH)

def test_lockf_exclusive(self):

self.f = open(TESTFN, 'wb+')

cmd = fcntl.LOCK_EX | fcntl.LOCK_NB

def try_lockf_on_other_process():

self.assertRaises(BlockingIOError, fcntl.lockf, self.f, cmd)

fcntl.lockf(self.f, cmd)

p = Process(target=try_lockf_on_other_process)

p.start()

p.join()

fcntl.lockf(self.f, fcntl.LOCK_UN)

self.assertEqual(p.exitcode, 0)

def test_lockf_share(self):

self.f = open(TESTFN, 'wb+')

cmd = fcntl.LOCK_SH | fcntl.LOCK_NB

def try_lockf_on_other_process():

fcntl.lockf(self.f, cmd)

fcntl.lockf(self.f, fcntl.LOCK_UN)

fcntl.lockf(self.f, cmd)

p = Process(target=try_lockf_on_other_process)

p.start()

p.join()

fcntl.lockf(self.f, fcntl.LOCK_UN)

self.assertEqual(p.exitcode, 0)

@cpython_only

def test_flock_overflow(self):

import _testcapi