Packages

snakepit

0.4.2
0.13.0 0.12.0 0.11.1 0.11.0 0.10.1 0.10.0 0.9.1 0.9.0 0.8.9 0.8.8 0.8.7 0.8.6 0.8.5 0.8.4 0.8.3 0.8.2 0.8.1 0.8.0 0.7.7 0.7.6 0.7.5 0.7.4 0.7.3 0.7.2 0.7.1 0.7.0 0.6.11 0.6.10 0.6.9 0.6.8 0.6.7 0.6.6 0.6.5 0.6.4 0.6.3 0.6.2 0.6.1 0.6.0 0.5.1 0.5.0 0.4.3 0.4.2 0.4.1 0.4.0 0.3.3 0.3.2 0.3.1 0.3.0 0.2.1 0.2.0 0.1.2 0.1.1 0.1.0

High-performance pooler and session manager for external language integrations. Supports Python, Node.js, Ruby, and more with gRPC streaming, session management, and production-ready process cleanup.

HexDocs

Current section

Files

Jump to
Readme
53 Versions
3 Dependencies
Dependants
Files
Activity
snakepit priv python test_server.py
Raw

priv/python/test_server.py 

#!/usr/bin/env python3

"""

Test the gRPC server startup

"""



import sys

import os



# Add the package to Python path

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))



# Test imports first

try:

    from snakepit_bridge import snakepit_bridge_pb2

    from snakepit_bridge import snakepit_bridge_pb2_grpc

    print("✓ Successfully imported generated protobuf modules")

except ImportError as e:

    print(f"✗ Failed to import protobuf modules: {e}")

    sys.exit(1)



try:

    from snakepit_bridge.session_context import SessionContext

    from snakepit_bridge.serialization import TypeSerializer

    print("✓ Successfully imported bridge modules")

except ImportError as e:

    print(f"✗ Failed to import bridge modules: {e}")

    sys.exit(1)



# Test starting the server

import subprocess

import time

import signal



print("\nStarting gRPC server...")

proc = subprocess.Popen(

    [sys.executable, "grpc_server.py", "--port", "50051"],

    stdout=subprocess.PIPE,

    stderr=subprocess.PIPE,

    text=True,

    bufsize=1,

    universal_newlines=True

)



# Wait for server to start

ready = False

start_time = time.time()

while time.time() - start_time < 10:

    line = proc.stdout.readline()

    if line:

        print(f"Server output: {line.strip()}")

        if "GRPC_READY:" in line:

            ready = True

            break

    

    # Check if process died

    if proc.poll() is not None:

        stderr = proc.stderr.read()

        print(f"Server died with error:\n{stderr}")

        sys.exit(1)



if ready:

    print("\n✓ Server started successfully!")

    print("Shutting down server...")

    proc.send_signal(signal.SIGTERM)

    proc.wait(timeout=5)

    print("✓ Server shutdown cleanly")

else:

    print("\n✗ Server failed to start within timeout")

    proc.kill()

    sys.exit(1)