62 lines
2.5 KiB
Python
62 lines
2.5 KiB
Python
# hobo - portal to configure and deploy applications
|
|
# Copyright (C) 2022 Entr'ouvert
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify it
|
|
# under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import hashlib
|
|
import os
|
|
import socket
|
|
import struct
|
|
from contextlib import closing
|
|
|
|
|
|
def get_safe_db_name(max_length=53):
|
|
"""
|
|
PostgreSQL database name limit is 63 characters, which can become
|
|
an issue during testing, because we need to build a unique
|
|
database name using the branch name and tox env.
|
|
|
|
Also, when running tests in parallel through `tox -p`,
|
|
pytest django append the tox env name automatically
|
|
through a fixture so we have to skip this step.
|
|
"""
|
|
if not os.environ.get('TOX_ENV_NAME'):
|
|
return 'hobo-test'
|
|
BRANCH_NAME = os.environ.get('BRANCH_NAME', '').replace('/', '-')[:15]
|
|
parts = [BRANCH_NAME]
|
|
if not os.environ.get("TOX_PARALLEL_ENV"):
|
|
# when we're in parallel mode, pytest-django will do this
|
|
# for us at a later point
|
|
parts.append(os.environ.get('TOX_ENV_NAME'))
|
|
full_db_name = '_'.join(parts)
|
|
if len(full_db_name) < max_length:
|
|
return full_db_name
|
|
hashcode_length = 8
|
|
hashcode = hashlib.md5(full_db_name.encode()).hexdigest()[: hashcode_length - 2]
|
|
prefix_length = (max_length - hashcode_length) - (max_length - hashcode_length) // 2
|
|
suffix_length = (max_length - hashcode_length) // 2
|
|
assert hashcode_length + prefix_length + suffix_length == max_length
|
|
truncated_db_name = full_db_name[:prefix_length] + '_' + hashcode + '_' + full_db_name[-suffix_length:]
|
|
assert len(truncated_db_name) == max_length
|
|
return truncated_db_name
|
|
|
|
|
|
def find_free_port():
|
|
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
|
|
s.bind(('', 0))
|
|
# SO_LINGER (man 7 socket) l_onoff=1 l_linger=0, immediately release
|
|
# the port on closing of the socket
|
|
s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
|
|
return s.getsockname()[1]
|