Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py #!/usr/bin/env python # -- coding: utf-8 -- """unit pytests for :mod:`graypy` .. note:: These tests mock sending to Graylog, thus, do not require a local instance of Graylog to successfully run. """	212 B	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 helper.py #!/usr/bin/env python # -- coding: utf-8 -- """helper functions for testing graypy with mocks of python logging and Graylog services""" import logging MOCK_LOG_RECORD_NAME = "MOCK_LOG_RECORD" MOCK_LOG_RECORD = logging.LogRecord( MOCK_LOG_RECORD_NAME, logging.INFO, pathname=None, lineno=None, msg="Log message", args=(), exc_info=None, )	371 B	2025-05-25 12:26
📄 test_ExcludeFilter.py #!/usr/bin/env python # -- coding: utf-8 -- """pytests for :class:`graypy.rabbitmq.ExcludeFilter`""" import pytest from graypy import ExcludeFilter from tests.unit.helper import MOCK_LOG_RECORD_NAME, MOCK_LOG_RECORD @pytest.mark.parametrize("name", [None, ""]) def test_invalid_name(name): """Test constructing:class:`graypy.rabbitmq.ExcludeFilter` with a invalid ``name`` argument""" with pytest.raises(ValueError): ExcludeFilter(name) @pytest.mark.parametrize("name", ["foobar", ".", " "]) def test_valid_name(name): """Test constructing :class:`graypy.rabbitmq.ExcludeFilter` with a valid ``name`` argument""" exclude_filter = ExcludeFilter(name) assert exclude_filter assert name == exclude_filter.name assert len(name) == exclude_filter.nlen def test_non_filtering_record(): exclude_filter = ExcludeFilter("NOT" + MOCK_LOG_RECORD_NAME) assert exclude_filter.filter(MOCK_LOG_RECORD) assert MOCK_LOG_RECORD.name != exclude_filter.name def test_filtering_record(): exclude_filter = ExcludeFilter(MOCK_LOG_RECORD_NAME) assert not exclude_filter.filter(MOCK_LOG_RECORD) assert MOCK_LOG_RECORD.name == exclude_filter.name	1.17 KB	2025-05-25 12:26
📄 test_GELFRabbitHandler.py #!/usr/bin/env python # -- coding: utf-8 -- """pytests for :class:`graypy.rabbitmq.GELFRabbitHandler`""" import json import pytest from graypy.rabbitmq import GELFRabbitHandler from graypy.handler import SYSLOG_LEVELS from tests.unit.helper import MOCK_LOG_RECORD def test_invalid_url(): """Test constructing :class:`graypy.rabbitmq.GELFRabbitHandler` with an invalid rabbitmq url""" with pytest.raises(ValueError): GELFRabbitHandler("BADURL") def test_valid_url(): """Test constructing :class:`graypy.rabbitmq.GELFRabbitHandler` with a valid rabbitmq url""" handler = GELFRabbitHandler("amqp://localhost") assert handler assert "amqp://localhost" == handler.url @pytest.mark.xfail(reason="rabbitmq service is not up") def test_socket_creation_failure(): """Test attempting to open a socket to a rabbitmq instance when no such service exists""" handler = GELFRabbitHandler("amqp://localhost") handler.makeSocket() def test_make_pickle(): handler = GELFRabbitHandler("amqp://localhost") pickle = json.loads(handler.makePickle(MOCK_LOG_RECORD)) assert "Log message" == pickle["short_message"] assert SYSLOG_LEVELS[MOCK_LOG_RECORD.levelno] == pickle["level"]	1.21 KB	2025-05-25 12:26
📄 test_chunking.py #!/usr/bin/env python # -- coding: utf-8 -- """pytests for various GELF UDP message chunkers""" import json import logging import struct import zlib import pytest from graypy.handler import ( GELFTruncatingChunker, GELFWarningChunker, BaseGELFChunker, BaseGELFHandler, SYSLOG_LEVELS, GELFChunkOverflowWarning, GELFTruncationFailureWarning, ) @pytest.mark.parametrize( "gelf_chunker", [BaseGELFChunker, GELFWarningChunker, GELFTruncatingChunker] ) def test_gelf_chunking(gelf_chunker): """Test various GELF chunkers""" message = b"12345" header = b"\x1e\x0f" chunks = list(gelf_chunker(chunk_size=2).chunk_message(message)) expected = [ (struct.pack("b", 0), struct.pack("b", 3), b"12"), (struct.pack("b", 1), struct.pack("b", 3), b"34"), (struct.pack("b", 2), struct.pack("b", 3), b"5"), ] assert len(chunks) == len(expected) for index, chunk in enumerate(chunks): expected_index, expected_chunks_count, expected_chunk = expected[index] assert header == chunk[:2] assert expected_index == chunk[10:11] assert expected_chunks_count == chunk[11:12] assert expected_chunk == chunk[12:] def rebuild_gelf_bytes_from_udp_chunks(chunks): gelf_bytes = b"" bsize = len(chunks[0]) for chunk in chunks: if len(chunk) < bsize: gelf_bytes += chunk[-(bsize - len(chunk)) :] else: gelf_bytes += chunk[((2 + struct.calcsize("QBB")) - len(chunk)) :] return gelf_bytes @pytest.mark.parametrize( "gelf_chunker", [BaseGELFChunker, GELFWarningChunker, GELFTruncatingChunker] ) def test_gelf_chunkers(gelf_chunker): message = BaseGELFHandler().makePickle( logging.LogRecord( "test_gelf_chunkers", logging.INFO, None, None, "1" * 10, None, None ) ) chunks = list(gelf_chunker(chunk_size=2).chunk_message(message)) assert len(chunks) <= 128 @pytest.mark.parametrize( "gelf_chunker", [BaseGELFChunker, GELFWarningChunker, GELFTruncatingChunker] ) def test_gelf_chunkers_overflow(gelf_chunker): message = BaseGELFHandler().makePickle( logging.LogRecord( "test_gelf_chunkers_overflow", logging.INFO, None, None, "1" * 1000, None, None, ) ) chunks = list(gelf_chunker(chunk_size=1).chunk_message(message)) assert len(chunks) <= 128 def test_chunk_overflow_truncate_uncompressed(): message = BaseGELFHandler(compress=False).makePickle( logging.LogRecord( "test_chunk_overflow_truncate_uncompressed", logging.INFO, None, None, "1" * 1000, None, None, ) ) with pytest.warns(GELFChunkOverflowWarning): chunks = list( GELFTruncatingChunker(chunk_size=2, compress=False).chunk_message(message) ) assert len(chunks) <= 128 payload = rebuild_gelf_bytes_from_udp_chunks(chunks).decode("UTF-8") glef_json = json.loads(payload) assert glef_json["_chunk_overflow"] is True assert glef_json["short_message"] in "1" * 1000 assert glef_json["level"] == SYSLOG_LEVELS.get(logging.ERROR, logging.ERROR) def test_chunk_overflow_truncate_compressed(): message = BaseGELFHandler(compress=True).makePickle( logging.LogRecord( "test_chunk_overflow_truncate_compressed", logging.INFO, None, None, "123412345" * 5000, None, None, ) ) with pytest.warns(GELFChunkOverflowWarning): chunks = list( GELFTruncatingChunker(chunk_size=2, compress=True).chunk_message(message) ) assert len(chunks) <= 128 payload = zlib.decompress(rebuild_gelf_bytes_from_udp_chunks(chunks)).decode( "UTF-8" ) glef_json = json.loads(payload) assert glef_json["_chunk_overflow"] is True assert glef_json["short_message"] in "123412345" * 5000 assert glef_json["level"] == SYSLOG_LEVELS.get(logging.ERROR, logging.ERROR) def test_chunk_overflow_truncate_fail(): message = BaseGELFHandler().makePickle( logging.LogRecord( "test_chunk_overflow_truncate_fail", logging.INFO, None, None, "1" * 1000, None, None, ) ) with pytest.warns(GELFTruncationFailureWarning): list(GELFTruncatingChunker(1).chunk_message(message)) def test_chunk_overflow_truncate_fail_large_inherited_field(): message = BaseGELFHandler( facility="this is a really long facility" * 5000 ).makePickle( logging.LogRecord( "test_chunk_overflow_truncate_fail", logging.INFO, None, None, "reasonable message", None, None, ) ) with pytest.warns(GELFTruncationFailureWarning): list(GELFTruncatingChunker(2).chunk_message(message))	4.96 KB	2025-05-25 12:26
📄 test_handler.py #!/usr/bin/env python # -- coding: utf-8 -- """pytests for the formatting and construction of GELF logs by the graypy logging handlers .. note:: These tests mock sending to Graylog and do not require an active graylog instance to operate. """ import datetime import json import logging import socket import sys import zlib import mock import pytest from graypy.handler import BaseGELFHandler, GELFHTTPHandler, GELFTLSHandler from tests.helper import handler, logger, formatted_logger from tests.unit.helper import MOCK_LOG_RECORD, MOCK_LOG_RECORD_NAME UNICODE_REPLACEMENT = u"\ufffd" class TestClass(object): def __repr__(self): return "<TestClass>" @pytest.yield_fixture def mock_send(handler): try: with mock.patch.object(handler, "send") as mock_send: yield mock_send except Exception: with mock.patch.object(handler, "emit") as mock_send: yield mock_send def get_mock_send_arg(mock_send): assert mock_send.call_args_list != [] [[[arg], _]] = mock_send.call_args_list # TODO: this is inaccurate solution for mocking non-send handlers if isinstance(arg, logging.LogRecord): return json.loads( BaseGELFHandler(compress=False).makePickle(arg).decode("utf-8") ) try: return json.loads(zlib.decompress(arg).decode("utf-8")) except zlib.error: # we have a uncompress message try: return json.loads(arg.decode("utf-8")) except Exception: # that is null terminated return json.loads(arg[:-1].decode("utf-8")) @pytest.mark.parametrize( "message,expected", [ (u"\u20AC", u"\u20AC"), (u"\u20AC".encode("utf-8"), u"\u20AC"), (b"\xc3", UNICODE_REPLACEMENT), (["a", b"\xc3"], ["a", UNICODE_REPLACEMENT]), ], ) def test_pack(message, expected): assert expected == json.loads( BaseGELFHandler._pack_gelf_dict(message).decode("utf-8") ) def test_manual_exc_info_handler(logger, mock_send): """Check that a the ``full_message`` traceback info is passed when the ``exc_info=1`` flag is given within a log message""" try: raise SyntaxError("Syntax error") except SyntaxError: logger.error("Failed", exc_info=1) arg = get_mock_send_arg(mock_send) assert "Failed" == arg["short_message"] assert arg["full_message"].startswith("Traceback (most recent call last):") # GELFHTTPHandler mocking does not complete the stacktrace # thus a missing \n assert arg["full_message"].endswith("SyntaxError: Syntax error") or arg[ "full_message" ].endswith("SyntaxError: Syntax error\n") def test_normal_exception_handler(logger, mock_send): try: raise SyntaxError("Syntax error") except SyntaxError: logger.exception("Failed") arg = get_mock_send_arg(mock_send) assert "Failed" == arg["short_message"] assert arg["full_message"].startswith("Traceback (most recent call last):") # GELFHTTPHandler mocking does not complete the stacktrace # thus a missing \n assert arg["full_message"].endswith("SyntaxError: Syntax error") or arg[ "full_message" ].endswith("SyntaxError: Syntax error\n") def test_unicode(logger, mock_send): logger.error(u"Mensaje de registro espa\xf1ol") arg = get_mock_send_arg(mock_send) assert u"Mensaje de registro espa\xf1ol" == arg["short_message"] @pytest.mark.skipif(sys.version_info[0] >= 3, reason="python2 only") def test_broken_unicode_python2(logger, mock_send): # py3 record.getMessage() returns a binary string here # which is safely converted to unicode during the sanitization # process logger.error(b"Broken \xde log message") decoded = get_mock_send_arg(mock_send) assert u"Broken %s log message" % UNICODE_REPLACEMENT == decoded["short_message"] @pytest.mark.skipif(sys.version_info[0] < 3, reason="python3 only") def test_broken_unicode_python3(logger, mock_send): # py3 record.getMessage() returns somewhat broken "b"foo"" if the # message string is not a string, but a binary object: b"foo" logger.error(b"Broken \xde log message") decoded = get_mock_send_arg(mock_send) assert "b'Broken \\xde log message'" == decoded["short_message"] def test_extra_field(logger, mock_send): logger.error("Log message", extra={"foo": "bar"}) decoded = get_mock_send_arg(mock_send) assert "Log message" == decoded["short_message"] assert "bar" == decoded["_foo"] def test_list(logger, mock_send): logger.error("Log message", extra={"foo": ["bar", "baz"]}) decoded = get_mock_send_arg(mock_send) assert "Log message" == decoded["short_message"] assert ["bar", "baz"] == decoded["_foo"] def test_arbitrary_object(logger, mock_send): logger.error("Log message", extra={"foo": TestClass()}) decoded = get_mock_send_arg(mock_send) assert "Log message" == decoded["short_message"] assert "<TestClass>" == decoded["_foo"] def test_message_to_pickle_serializes_datetime_objects_instead_of_blindly_repring_them( logger, mock_send ): timestamp = datetime.datetime(2001, 2, 3, 4, 5, 6, 7) logger.error("Log message", extra={"ts": timestamp}) decoded = get_mock_send_arg(mock_send) assert "datetime.datetime" not in decoded["_ts"] assert timestamp.isoformat() == decoded["_ts"] def test_status_field_issue(logger, mock_send): logger.error("Log message", extra={"status": "OK"}) decoded = get_mock_send_arg(mock_send) assert "Log message" == decoded["short_message"] assert "OK" == decoded["_status"] def test_add_level_name(): gelf_dict = dict() BaseGELFHandler._add_level_names(gelf_dict, MOCK_LOG_RECORD) assert "INFO" == gelf_dict["level_name"] def test_resolve_host(): """Test all posible resolutions of :meth:`BaseGELFHandler._resolve_host`""" assert socket.gethostname() == BaseGELFHandler._resolve_host(False, None) assert socket.getfqdn() == BaseGELFHandler._resolve_host(True, None) assert socket.getfqdn() == BaseGELFHandler._resolve_host(True, "localhost") assert "localhost" == BaseGELFHandler._resolve_host(False, "localhost") assert "" == BaseGELFHandler._resolve_host(False, "") def test_set_custom_facility(): gelf_dict = dict() facility = "test facility" BaseGELFHandler._set_custom_facility(gelf_dict, facility, MOCK_LOG_RECORD) assert MOCK_LOG_RECORD_NAME == gelf_dict["_logger"] assert "test facility" == gelf_dict["facility"] def test_formatted_logger(formatted_logger, mock_send): """Test the ability to set and modify the graypy handler's :class:`logging.Formatter` and have the resultant ``short_message`` be formatted by the set :class:`logging.Formatter`""" for handler in formatted_logger.handlers: if isinstance(handler, GELFHTTPHandler): pytest.skip("formatting not mocked for GELFHTTPHandler") formatted_logger.error("Log message") decoded = get_mock_send_arg(mock_send) assert "ERROR : Log message" == decoded["short_message"] def test_invalid_fqdn_localhost(): """Test constructing :class:`graypy.handler.BaseGELFHandler` with specifying conflicting arguments ``fqdn`` and ``localname``""" with pytest.raises(ValueError): BaseGELFHandler(fqdn=True, localname="localhost") def test_invalid_ca_certs(): """Test constructing :class:`graypy.handler.GELFTLSHandler` with incorrect arguments specifying server ca cert verification""" with pytest.raises(ValueError): GELFTLSHandler("127.0.0.1", validate=True) def test_invalid_client_certs(): """Test constructing :class:`graypy.handler.GELFTLSHandler` with incorrect arguments specifying client cert/key verification""" with pytest.raises(ValueError): # missing client cert GELFTLSHandler("127.0.0.1", keyfile="/dev/null")	7.72 KB	2025-05-25 12:26

Terminal