# Path: tests/_adapt.py
# IS_GAE = "datastore" in DEFAULT_URI
# 
# Path: pydal/_compat.py
# def to_bytes(obj, charset="utf-8", errors="strict"):
#     if obj is None:
#         return None
#     if isinstance(obj, (bytes, bytearray, buffer)):
#         return bytes(obj)
#     if isinstance(obj, unicode):
#         return obj.encode(charset, errors)
#     raise TypeError("Expected bytes")
# 
# Path: pydal/contrib/portalocker.py
# def lock(file, flags):
#     hfile = msvcrt.get_osfhandle(file.fileno())
#     overlapped = OVERLAPPED()
#     LockFileEx(hfile, flags, 0, 0, 0xFFFF0000, ctypes.byref(overlapped))
# 
# def unlock(file):
#     hfile = msvcrt.get_osfhandle(file.fileno())
#     overlapped = OVERLAPPED()
#     UnlockFileEx(hfile, 0, 0, 0xFFFF0000, ctypes.byref(overlapped))
# 
# def read_locked(filename):
#     fp = LockedFile(filename, "rb")
#     data = fp.read()
#     fp.close()
#     return data
# 
# def write_locked(filename, data):
#     fp = LockedFile(filename, "wb")
#     data = fp.write(data)
#     fp.close()
# 
# Path: pydal/contrib/portalocker.py
# class LockedFile(object):
#     def __init__(self, filename, mode="rb"):
#         self.filename = filename
#         self.mode = mode
#         self.file = None
#         if "r" in mode:
#             self.file = open_file(filename, mode)
#             lock(self.file, LOCK_SH)
#         elif "w" in mode or "a" in mode:
#             self.file = open_file(filename, mode.replace("w", "a"))
#             lock(self.file, LOCK_EX)
#             if "a" not in mode:
#                 self.file.seek(0)
#                 self.file.truncate(0)
#         else:
#             raise RuntimeError("invalid LockedFile(...,mode)")
# 
#     def read(self, size=None):
#         return self.file.read() if size is None else self.file.read(size)
# 
#     def readinto(self, b):
#         b[:] = self.file.read()
# 
#     def readline(self):
#         return self.file.readline()
# 
#     def readlines(self):
#         return self.file.readlines()
# 
#     def write(self, data):
#         self.file.write(data)
#         self.file.flush()
# 
#     def close(self):
#         if self.file is not None:
#             unlock(self.file)
#             self.file.close()
#             self.file = None
# 
#     def __del__(self):
#         if self.file is not None:
#             self.close()
# 
# LOCK_EX = 0x2  # LOCKFILE_EXCLUSIVE_LOCK


import os
import threading
import time
from ._compat import unittest
from ._adapt import IS_GAE
from pydal._compat import to_bytes
from pydal.contrib.portalocker import lock, unlock, read_locked, write_locked
from pydal.contrib.portalocker import LockedFile, LOCK_EX
        content = read_locked("test.txt")
        end = int(time.time())
        t1.join()
        f.close()
        # it took at least 2 seconds to read
        # although nothing is there until .close()
        self.assertTrue(end - start >= 2)
        self.assertEqual(content, to_bytes(""))
        content = read_locked("test.txt")
        self.assertEqual(content, to_bytes("test ok"))

        f = LockedFile("test.txt", mode="wb")
        f.write(to_bytes("test ok"))
        t1 = threading.Thread(target=worker2, args=(f,))
        t1.start()
        start = int(time.time())
        content = read_locked("test.txt")
        end = int(time.time())
        t1.join()
        # it took at least 2 seconds to read
        # content is there because we called close()
        self.assertTrue(end - start >= 2)
        self.assertEqual(content, to_bytes("test ok"))

    @unittest.skipIf(IS_GAE, "GAE has no locks")
    def test_read_locked(self):
        def worker(fh):
            time.sleep(2)
            fh.close()


# Path: tests/_adapt.py
# IS_GAE = "datastore" in DEFAULT_URI
# 
# Path: pydal/_compat.py
# def to_bytes(obj, charset="utf-8", errors="strict"):
#     if obj is None:
#         return None
#     if isinstance(obj, (bytes, bytearray, buffer)):
#         return bytes(obj)
#     if isinstance(obj, unicode):
#         return obj.encode(charset, errors)
#     raise TypeError("Expected bytes")
# 
# Path: pydal/contrib/portalocker.py
# def lock(file, flags):
#     hfile = msvcrt.get_osfhandle(file.fileno())
#     overlapped = OVERLAPPED()
#     LockFileEx(hfile, flags, 0, 0, 0xFFFF0000, ctypes.byref(overlapped))
# 
# def unlock(file):
#     hfile = msvcrt.get_osfhandle(file.fileno())
#     overlapped = OVERLAPPED()
#     UnlockFileEx(hfile, 0, 0, 0xFFFF0000, ctypes.byref(overlapped))
# 
# def read_locked(filename):
#     fp = LockedFile(filename, "rb")
#     data = fp.read()
#     fp.close()
#     return data
# 
# def write_locked(filename, data):
#     fp = LockedFile(filename, "wb")
#     data = fp.write(data)
#     fp.close()
# 
# Path: pydal/contrib/portalocker.py
# class LockedFile(object):
#     def __init__(self, filename, mode="rb"):
#         self.filename = filename
#         self.mode = mode
#         self.file = None
#         if "r" in mode:
#             self.file = open_file(filename, mode)
#             lock(self.file, LOCK_SH)
#         elif "w" in mode or "a" in mode:
#             self.file = open_file(filename, mode.replace("w", "a"))
#             lock(self.file, LOCK_EX)
#             if "a" not in mode:
#                 self.file.seek(0)
#                 self.file.truncate(0)
#         else:
#             raise RuntimeError("invalid LockedFile(...,mode)")
# 
#     def read(self, size=None):
#         return self.file.read() if size is None else self.file.read(size)
# 
#     def readinto(self, b):
#         b[:] = self.file.read()
# 
#     def readline(self):
#         return self.file.readline()
# 
#     def readlines(self):
#         return self.file.readlines()
# 
#     def write(self, data):
#         self.file.write(data)
#         self.file.flush()
# 
#     def close(self):
#         if self.file is not None:
#             unlock(self.file)
#             self.file.close()
#             self.file = None
# 
#     def __del__(self):
#         if self.file is not None:
#             self.close()
# 
# LOCK_EX = 0x2  # LOCKFILE_EXCLUSIVE_LOCK


# Path: tests/contribs.py
import os
import threading
import time
from ._compat import unittest
from ._adapt import IS_GAE
from pydal._compat import to_bytes
from pydal.contrib.portalocker import lock, unlock, read_locked, write_locked
from pydal.contrib.portalocker import LockedFile, LOCK_EX

        content = read_locked("test.txt")
        end = int(time.time())
        t1.join()
        f.close()
        # it took at least 2 seconds to read
        # although nothing is there until .close()
        self.assertTrue(end - start >= 2)
        self.assertEqual(content, to_bytes(""))
        content = read_locked("test.txt")
        self.assertEqual(content, to_bytes("test ok"))

        f = LockedFile("test.txt", mode="wb")
        f.write(to_bytes("test ok"))
        t1 = threading.Thread(target=worker2, args=(f,))
        t1.start()
        start = int(time.time())
        content = read_locked("test.txt")
        end = int(time.time())
        t1.join()
        # it took at least 2 seconds to read
        # content is there because we called close()
        self.assertTrue(end - start >= 2)
        self.assertEqual(content, to_bytes("test ok"))

    @unittest.skipIf(IS_GAE, "GAE has no locks")
    def test_read_locked(self):
        def worker(fh):
            time.sleep(2)
            fh.close()


        f = LockedFile("test.txt", mode="wb")