# -*- coding: utf-8 -*-
# :Project:   python-rapidjson -- Refs leaks tests
# :Author:    Lele Gaifax <lele@metapensiero.it>
# :License:   MIT License
# :Copyright: © 2017, 2018, 2020, 2022, 2025 Lele Gaifax
#

# NB: this is a simplistic test that uses sys.gettotalrefcount(), available
# when the interpreter is built --with-pydebug, that tries to assert that
# repeated calls to dumps() and loads() does not leak object references.
# Since it's not an exact science, it should be taken with a grain of salt.

import datetime
import io
import sys
import uuid

import pytest

import rapidjson as rj


class Foo:
    def __init__(self, foo):
        self.foo = foo


def hook(d):
    if 'foo' in d:
        return Foo(d['foo'])
    return d


def default(obj):
    return {'foo': obj.foo}


# Empirical tolerance used to test the refcount growing
THRESHOLD = 7


plain_string = 'plain-string'
bigint = 123456789
pi = 3.1415926
right_now = datetime.datetime.now()
date = right_now.date()
time = right_now.time()
nil = uuid.UUID(int=0)
foo = Foo('foo')
array = [dict(index=j, array=[dict(index=i,
                                   plain_string=plain_string,
                                   bigint=bigint,
                                   pi=pi,
                                   right_now=right_now,
                                   date=date,
                                   time=time,
                                   nil=nil,
                                   foos=[foo, foo, foo, foo, foo])
                              for i in range(10)])
         for j in range(10)]

NO_OPTION = {}
DATETIMES = {'datetime_mode': rj.DM_ISO8601}
UUIDS = {'uuid_mode': rj.UM_CANONICAL}
FOOS_DUMP = {'default': default}
FOOS_LOAD = {'object_hook': hook}
ARRAY_DUMP = {'datetime_mode': rj.DM_ISO8601,
              'uuid_mode': rj.UM_CANONICAL,
              'default': default}
ARRAY_LOAD = {'datetime_mode': rj.DM_ISO8601,
              'uuid_mode': rj.UM_CANONICAL,
              'object_hook': hook}

@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
@pytest.mark.parametrize('value,dumps_options,loads_options', [
    ( plain_string, NO_OPTION, NO_OPTION ),
    ( bigint, NO_OPTION, NO_OPTION ),
    ( pi, NO_OPTION, NO_OPTION ),
    ( rj.RawJSON(' "foo" '), NO_OPTION, NO_OPTION),
    ( right_now, DATETIMES, DATETIMES ),
    ( date, DATETIMES, DATETIMES ),
    ( time, DATETIMES, DATETIMES ),
    ( nil, UUIDS, UUIDS ),
    ( foo, FOOS_DUMP, FOOS_LOAD ),
    ( array, ARRAY_DUMP, ARRAY_LOAD ),
])
def test_leaks(value, dumps_options, loads_options):
    rc0 = sys.gettotalrefcount()
    for _ in range(1000):
        asjson = rj.dumps(value, **dumps_options)
        aspython = rj.loads(asjson, **loads_options)
        del asjson
        del aspython
    rc1 = sys.gettotalrefcount()
    assert (rc1 - rc0) < THRESHOLD

    rc0 = sys.gettotalrefcount()
    for _ in range(1000):
        stream = io.BytesIO()
        none = rj.dump(value, stream, **dumps_options)
        stream.seek(0)
        aspython = rj.load(stream, **loads_options)
        del none
        del aspython
        del stream
    rc1 = sys.gettotalrefcount()
    assert (rc1 - rc0) < THRESHOLD


@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
def test_dump_iterator_leaks():
    rc0 = sys.gettotalrefcount()
    value = iter(range(10000))
    asjson = rj.dumps(value)
    del asjson
    del value
    rc1 = sys.gettotalrefcount()
    assert (rc1 - rc0) < THRESHOLD


@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
def test_decoder_call_leaks():
    class MyDecoder(rj.Decoder):
        def start_object(self):
            return {}
        def end_object(self, obj):
            return obj
        def end_array(self, array):
            return array
        def string(self, string):
            return string

    decoder = MyDecoder()
    rc0 = sys.gettotalrefcount()
    for _ in range(1000):
        value = decoder('["foo", {"foo": "bar"}]')
        del value
    rc1 = sys.gettotalrefcount()
    assert (rc1 - rc0) < THRESHOLD


@pytest.mark.parametrize('value', ['Foo', rj.RawJSON('Foo')])
@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
def test_encoder_call_leaks(value):
    class MyEncoder(rj.Encoder):
        def __init__(self, value):
            self.value = value

        def default(self, obj):
            return self.value

    class Foo:
        pass

    encoder = MyEncoder(value)
    foo = Foo()
    rc0 = sys.gettotalrefcount()
    for _ in range(1000):
        value = encoder(foo)
        del value
    rc1 = sys.gettotalrefcount()
    assert (rc1 - rc0) < THRESHOLD


@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
def test_rawjson_constructor():
    raw_json = rj.RawJSON('["foo", "bar"]')
    rc0 = sys.gettotalrefcount()
    for _ in range(1000):
        value = '"foobar"'
        raw_json.__init__(value)
        del value
    rc1 = sys.gettotalrefcount()
    assert (rc1 - rc0) < THRESHOLD


@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
def test_rawjson_new():
    rc0 = sys.gettotalrefcount()
    for _ in range(1000):
        raw_json = rj.RawJSON('["foo", "bar"]')
        del raw_json
    rc1 = sys.gettotalrefcount()
    assert (rc1 - rc0) < THRESHOLD


@pytest.mark.skipif(not hasattr(sys, 'gettotalrefcount'), reason='Non-debug Python')
def test_invalid_json_load_leaks():
    # See issue #148
    value = '{"a":{"b":}}'
    rc0 = sys.gettotalrefcount()
    for _ in range(1000):
        try:
            rj.loads(value)
        except rj.JSONDecodeError:
            pass
    del _
    rc1 = sys.gettotalrefcount()
    assert (rc1 - rc0) < THRESHOLD


def test_endarray_leak():
    # See issue #160
    value = '{"v": [1, 2]}'

    class Decoder1(rj.Decoder):
        pass

    class Decoder2(rj.Decoder):
        def end_array(self, seq):
            return list(seq)

    j1 = rj.loads(value)
    # Uhm, with Py 3.14 the refcount is 2...
    assert 2 <= sys.getrefcount(j1['v']) <= 3

    j2 = Decoder1()(value)
    assert 2 <= sys.getrefcount(j2['v']) <= 3

    j3 = Decoder2()(value)
    assert 2 <= sys.getrefcount(j3['v']) <= 3