13498621421

Committed 24 Feb 2025 12:59PM UTC coverage: 84.058% (+0.4%) from 83.706%

Build # 13498621421

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #545 from tlsfuzzer/ci-updates

add py3.13 to CI

Run Details

5070 of 6600 branches covered (76.82%)

Branch coverage included in aggregate %.

8 of 12 new or added lines in 1 file covered. (66.67%)

1 existing line in 1 file now uncovered.

12325 of 14094 relevant lines covered (87.45%)

47.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

54.17

/tlslite/basedb.py

# Authors: 
#   Trevor Perrin
#   Martin von Loewis - python 3 port
#
# See the LICENSE file for legal information regarding use of this file.

"""Base class for SharedKeyDB and VerifierDB."""

try:
    import anydbm
except ImportError:
    # Python 3
    import dbm as anydbm
import threading
import time
import logging

class BaseDB(object):
    def __init__(self, filename, type):
        self.type = type
        self.filename = filename
        if self.filename:
            self.db = None
        else:
            self.db = {}
        self.lock = threading.Lock()

    def create(self):
        """
        Create a new on-disk database.

        :raises anydbm.error: If there's a problem creating the database.
        """
        logger = logging.getLogger(__name__)

        if self.filename:
            logger.debug('server %s - create - will open db', time.time())
            self.db = anydbm.open(self.filename, "n") #raises anydbm.error
            logger.debug('server %s - create - setting type', time.time())
            self.db["--Reserved--type"] = self.type
            logger.debug('server %s - create - syncing', time.time())
            try:
                self.db.sync()
            except AttributeError:
                # some backends, like the py3.13 default sqlite , don't support
                # sync() method, so ignore it missing
                pass
            logger.debug('server %s - create - fun exit', time.time())
        else:
            logger.debug('server %s - create - using dict() as DB',
                         time.time())
            self.db = {}

    def open(self):
        """
        Open a pre-existing on-disk database.

        :raises anydbm.error: If there's a problem opening the database.
        :raises ValueError: If the database is not of the right type.
        """
        if not self.filename:
            raise ValueError("Can only open on-disk databases")
        self.db = anydbm.open(self.filename, "w") #raises anydbm.error
        try:
            if self.db["--Reserved--type"] != self.type:
                raise ValueError("Not a %s database" % self.type)
        except KeyError:
            raise ValueError("Not a recognized database")

    def __getitem__(self, username):
        if self.db == None:
            raise AssertionError("DB not open")

        self.lock.acquire()
        try:
            valueStr = self.db[username]
        finally:
            self.lock.release()

        return self._getItem(username, valueStr)

    def __setitem__(self, username, value):
        if self.db == None:
            raise AssertionError("DB not open")

        valueStr = self._setItem(username, value)

        self.lock.acquire()
        try:
            self.db[username] = valueStr
            if self.filename:
                try:
                    self.db.sync()
                except AttributeError:
                    pass
        finally:
            self.lock.release()

    def __delitem__(self, username):
        if self.db == None:
            raise AssertionError("DB not open")

        self.lock.acquire()
        try:
            del(self.db[username])
            if self.filename:
                try:
                    self.db.sync()
                except AttributeError:
                    pass
        finally:
            self.lock.release()

    def __contains__(self, username):
        """
        Check if the database contains the specified username.

        :param str username: The username to check for.

        :rtype: bool
        :returns: True if the database contains the username, False
            otherwise.
        """
        if self.db == None:
            raise AssertionError("DB not open")

        self.lock.acquire()
        try:
            return username in self.db
        finally:
            self.lock.release()

    def check(self, username, param):
        value = self.__getitem__(username)
        return self._checkItem(value, username, param)

    def keys(self):
        """
        Return a list of usernames in the database.

        :rtype: list
        :returns: The usernames in the database.
        """
        if self.db == None:
            raise AssertionError("DB not open")

        self.lock.acquire()
        try:
            usernames = self.db.keys()
        finally:
            self.lock.release()
        usernames = [u for u in usernames if not u.startswith("--Reserved--")]
        return usernames

1	# Authors:
2	# Trevor Perrin
3	# Martin von Loewis - python 3 port
4	#
5	# See the LICENSE file for legal information regarding use of this file.
6
7	"""Base class for SharedKeyDB and VerifierDB."""	27✔
8
9	try:	59✔
10	import anydbm	59✔
11	except ImportError:	51✔
12	# Python 3
13	import dbm as anydbm	51✔
14	import threading	59✔
15	import time	59✔
16	import logging	59✔
17
18	class BaseDB(object):	59✔
19	def __init__(self, filename, type):	59✔
20	self.type = type	59✔
21	self.filename = filename	59✔
22	if self.filename:	59✔
23	self.db = None	59✔
24	else:
25	self.db = {}	59✔
26	self.lock = threading.Lock()	59✔
27
28	def create(self):	59✔
29	"""
30	Create a new on-disk database.
31
32	:raises anydbm.error: If there's a problem creating the database.
33	"""
34	logger = logging.getLogger(__name__)	59✔
35
36	if self.filename:	59✔
37	logger.debug('server %s - create - will open db', time.time())	59✔
38	self.db = anydbm.open(self.filename, "n") #raises anydbm.error	59✔
39	logger.debug('server %s - create - setting type', time.time())	59✔
40	self.db["--Reserved--type"] = self.type	59✔
41	logger.debug('server %s - create - syncing', time.time())	59✔
42	try:	59✔
43	self.db.sync()	59✔
44	except AttributeError:	5✔
45	# some backends, like the py3.13 default sqlite , don't support
46	# sync() method, so ignore it missing
47	pass	5✔
48	logger.debug('server %s - create - fun exit', time.time())	59✔
49	else:
50	logger.debug('server %s - create - using dict() as DB',	59✔
51	time.time())
52	self.db = {}	59✔
53
54	def open(self):	59✔
55	"""
56	Open a pre-existing on-disk database.
57
58	:raises anydbm.error: If there's a problem opening the database.
59	:raises ValueError: If the database is not of the right type.
60	"""
61	if not self.filename:	×
62	raise ValueError("Can only open on-disk databases")	×
63	self.db = anydbm.open(self.filename, "w") #raises anydbm.error	×
64	try:	×
65	if self.db["--Reserved--type"] != self.type:	×
66	raise ValueError("Not a %s database" % self.type)	×
67	except KeyError:	×
68	raise ValueError("Not a recognized database")	×
69
70	def __getitem__(self, username):	59✔
71	if self.db == None:	59!
72	raise AssertionError("DB not open")	×
73
74	self.lock.acquire()	59✔
75	try:	59✔
76	valueStr = self.db[username]	59✔
77	finally:
78	self.lock.release()	59✔
79
80	return self._getItem(username, valueStr)	59✔
81
82	def __setitem__(self, username, value):	59✔
83	if self.db == None:	59!
84	raise AssertionError("DB not open")	×
85
86	valueStr = self._setItem(username, value)	59✔
87
88	self.lock.acquire()	59✔
89	try:	59✔
90	self.db[username] = valueStr	59✔
91	if self.filename:	59✔
92	try:	59✔
93	self.db.sync()	59✔
94	except AttributeError:	5✔
95	pass	5✔
96	finally:
97	self.lock.release()	59✔
98
99	def __delitem__(self, username):	59✔
100	if self.db == None:	×
101	raise AssertionError("DB not open")	×
102
103	self.lock.acquire()	×
104	try:	×
105	del(self.db[username])	×
106	if self.filename:	×
NEW 107	try:	×
NEW 108	self.db.sync()	×
NEW 109	except AttributeError:	×
NEW 110	pass	×
111	finally:
112	self.lock.release()	×
113
114	def __contains__(self, username):	59✔
115	"""
116	Check if the database contains the specified username.
117
118	:param str username: The username to check for.
119
120	:rtype: bool
121	:returns: True if the database contains the username, False
122	otherwise.
123	"""
124	if self.db == None:	×
125	raise AssertionError("DB not open")	×
126
127	self.lock.acquire()	×
128	try:	×
129	return username in self.db	×
130	finally:
131	self.lock.release()	×
132
133	def check(self, username, param):	59✔
134	value = self.__getitem__(username)	×
135	return self._checkItem(value, username, param)	×
136
137	def keys(self):	59✔
138	"""
139	Return a list of usernames in the database.
140
141	:rtype: list
142	:returns: The usernames in the database.
143	"""
144	if self.db == None:	×
145	raise AssertionError("DB not open")	×
146
147	self.lock.acquire()	×
148	try:	×
149	usernames = self.db.keys()	×
150	finally:
151	self.lock.release()	×
152	usernames = [u for u in usernames if not u.startswith("--Reserved--")]	×
153	return usernames	×

tlsfuzzer / tlslite-ng / 13498621421

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous