# Path: srsly/_msgpack_api.py
# def read_msgpack(path: FilePath, use_list: bool = True) -> JSONOutputBin:
#     """Load a msgpack file.
# 
#     location (FilePath): The file path.
#     use_list (bool): Don't use tuples instead of lists. Can make
#         deserialization slower.
#     RETURNS (JSONOutputBin): The loaded and deserialized content.
#     """
#     file_path = force_path(path)
#     with file_path.open("rb") as f:
#         # msgpack-python docs suggest disabling gc before unpacking large messages
#         gc.disable()
#         msg = msgpack.load(f, raw=False, use_list=use_list)
#         gc.enable()
#         return msg
# 
# def write_msgpack(path: FilePath, data: JSONInputBin) -> None:
#     """Create a msgpack file and dump contents.
# 
#     location (FilePath): The file path.
#     data (JSONInputBin): The data to serialize.
#     """
#     file_path = force_path(path, require_exists=False)
#     with file_path.open("wb") as f:
#         msgpack.dump(data, f, use_bin_type=True)
# 
# Path: srsly/_msgpack_api.py
# def msgpack_loads(data: bytes, use_list: bool = True) -> JSONOutputBin:
#     """Deserialize msgpack bytes to a Python object.
# 
#     data (bytes): The data to deserialize.
#     use_list (bool): Don't use tuples instead of lists. Can make
#         deserialization slower.
#     RETURNS: The deserialized Python object.
#     """
#     # msgpack-python docs suggest disabling gc before unpacking large messages
#     gc.disable()
#     msg = msgpack.loads(data, raw=False, use_list=use_list)
#     gc.enable()
#     return msg
# 
# def msgpack_dumps(data: JSONInputBin) -> bytes:
#     """Serialize an object to a msgpack byte string.
# 
#     data: The data to serialize.
#     RETURNS (bytes): The serialized bytes.
#     """
#     return msgpack.dumps(data, use_bin_type=True)
# 
# Path: srsly/_msgpack_api.py
# def msgpack_dumps(data: JSONInputBin) -> bytes:
# def msgpack_loads(data: bytes, use_list: bool = True) -> JSONOutputBin:
# def write_msgpack(path: FilePath, data: JSONInputBin) -> None:
# def read_msgpack(path: FilePath, use_list: bool = True) -> JSONOutputBin:
# 
# Path: srsly/tests/util.py
# @contextmanager
# def make_tempdir(files={}, mode="w"):
#     temp_dir_str = tempfile.mkdtemp()
#     temp_dir = Path(temp_dir_str)
#     for name, content in files.items():
#         path = temp_dir / name
#         with path.open(mode) as file_:
#             file_.write(content)
#     yield temp_dir
#     shutil.rmtree(temp_dir_str)


import pytest
import datetime
import numpy
from pathlib import Path
from mock import patch
from .._msgpack_api import read_msgpack, write_msgpack
from .._msgpack_api import msgpack_loads, msgpack_dumps
from .._msgpack_api import msgpack_encoders, msgpack_decoders
from .util import make_tempdir
def test_msgpack_without_numpy():
    """Test that msgpack works without numpy and raises correct errors (e.g.
    when serializing datetime objects, the error should be msgpack's TypeError,
    not a "'np' is not defined error")."""
    with pytest.raises(TypeError):
        msgpack_loads(msgpack_dumps(datetime.datetime.now()))


def test_msgpack_custom_encoder_decoder():
    class CustomObject:
        def __init__(self, value):
            self.value = value

    def serialize_obj(obj, chain=None):
        if isinstance(obj, CustomObject):
            return {"__custom__": obj.value}
        return obj if chain is None else chain(obj)

    def deserialize_obj(obj, chain=None):
        if "__custom__" in obj:
            return CustomObject(obj["__custom__"])
        return obj if chain is None else chain(obj)

    data = {"a": 123, "b": CustomObject({"foo": "bar"})}
    with pytest.raises(TypeError):
        msgpack_dumps(data)

    # Register custom encoders/decoders to handle CustomObject
    msgpack_encoders.register("custom_object", func=serialize_obj)
    msgpack_decoders.register("custom_object", func=deserialize_obj)

# Path: srsly/_msgpack_api.py
# def read_msgpack(path: FilePath, use_list: bool = True) -> JSONOutputBin:
#     """Load a msgpack file.
# 
#     location (FilePath): The file path.
#     use_list (bool): Don't use tuples instead of lists. Can make
#         deserialization slower.
#     RETURNS (JSONOutputBin): The loaded and deserialized content.
#     """
#     file_path = force_path(path)
#     with file_path.open("rb") as f:
#         # msgpack-python docs suggest disabling gc before unpacking large messages
#         gc.disable()
#         msg = msgpack.load(f, raw=False, use_list=use_list)
#         gc.enable()
#         return msg
# 
# def write_msgpack(path: FilePath, data: JSONInputBin) -> None:
#     """Create a msgpack file and dump contents.
# 
#     location (FilePath): The file path.
#     data (JSONInputBin): The data to serialize.
#     """
#     file_path = force_path(path, require_exists=False)
#     with file_path.open("wb") as f:
#         msgpack.dump(data, f, use_bin_type=True)
# 
# Path: srsly/_msgpack_api.py
# def msgpack_loads(data: bytes, use_list: bool = True) -> JSONOutputBin:
#     """Deserialize msgpack bytes to a Python object.
# 
#     data (bytes): The data to deserialize.
#     use_list (bool): Don't use tuples instead of lists. Can make
#         deserialization slower.
#     RETURNS: The deserialized Python object.
#     """
#     # msgpack-python docs suggest disabling gc before unpacking large messages
#     gc.disable()
#     msg = msgpack.loads(data, raw=False, use_list=use_list)
#     gc.enable()
#     return msg
# 
# def msgpack_dumps(data: JSONInputBin) -> bytes:
#     """Serialize an object to a msgpack byte string.
# 
#     data: The data to serialize.
#     RETURNS (bytes): The serialized bytes.
#     """
#     return msgpack.dumps(data, use_bin_type=True)
# 
# Path: srsly/_msgpack_api.py
# def msgpack_dumps(data: JSONInputBin) -> bytes:
# def msgpack_loads(data: bytes, use_list: bool = True) -> JSONOutputBin:
# def write_msgpack(path: FilePath, data: JSONInputBin) -> None:
# def read_msgpack(path: FilePath, use_list: bool = True) -> JSONOutputBin:
# 
# Path: srsly/tests/util.py
# @contextmanager
# def make_tempdir(files={}, mode="w"):
#     temp_dir_str = tempfile.mkdtemp()
#     temp_dir = Path(temp_dir_str)
#     for name, content in files.items():
#         path = temp_dir / name
#         with path.open(mode) as file_:
#             file_.write(content)
#     yield temp_dir
#     shutil.rmtree(temp_dir_str)


# Path: srsly/tests/test_msgpack_api.py
import pytest
import datetime
import numpy
from pathlib import Path
from mock import patch
from .._msgpack_api import read_msgpack, write_msgpack
from .._msgpack_api import msgpack_loads, msgpack_dumps
from .._msgpack_api import msgpack_encoders, msgpack_decoders
from .util import make_tempdir

def test_msgpack_without_numpy():
    """Test that msgpack works without numpy and raises correct errors (e.g.
    when serializing datetime objects, the error should be msgpack's TypeError,
    not a "'np' is not defined error")."""
    with pytest.raises(TypeError):
        msgpack_loads(msgpack_dumps(datetime.datetime.now()))


def test_msgpack_custom_encoder_decoder():
    class CustomObject:
        def __init__(self, value):
            self.value = value

    def serialize_obj(obj, chain=None):
        if isinstance(obj, CustomObject):
            return {"__custom__": obj.value}
        return obj if chain is None else chain(obj)

    def deserialize_obj(obj, chain=None):
        if "__custom__" in obj:
            return CustomObject(obj["__custom__"])
        return obj if chain is None else chain(obj)

    data = {"a": 123, "b": CustomObject({"foo": "bar"})}
    with pytest.raises(TypeError):
        msgpack_dumps(data)

    # Register custom encoders/decoders to handle CustomObject
    msgpack_encoders.register("custom_object", func=serialize_obj)
    msgpack_decoders.register("custom_object", func=deserialize_obj)

    bytes_data = msgpack_dumps(data)