Pity API Automation Testing Platform

Follow these steps to set up the pity API testing tool on your local machine:

1. Clone the repository:

   git clone https://github.com/wuranxu/pity
   cd pity

2. Install Python dependencies: You can use standard pip or alternative mirrors like Douban or Tsinghua if needed.

   pip install -r requirements.txt

3. Set up infrastructure:

   • Install and start a Redis instance.
   • Install and start a MySQL instance.

4. Configure the application: Edit the config.py file to provide the correct connection information for your Redis and MySQL instances.

5. Start the server:

   python pity.py

6. Access the UI: Open your browser and navigate to http://localhost:7777.

Note on Registration: The first user to register will be granted ADMIN privileges.

$ git clone https://github.com/wuranxu/pity
$ cd pity
$ pip install -r requirements.txt
$ python pity.py

To develop on Pity locally, follow these steps to set up the environment and run the server:

1. Clone the repository:

   git clone https://github.com/wuranxu/pity
   cd pity

2. Install dependencies (you can use Douban or Tsinghua mirrors for faster downloads):

   pip install -r requirements.txt

3. Start Infrastructure: Manually install and start redis and mysql.
4. Configure Environment: Edit conf/dev.env to provide the correct connection information for your MySQL and Redis instances. Note: Redis is highly recommended to prevent duplicate execution of scheduled tasks, as Pity uses Redis to implement distributed locks.
5. Run the server:

   python pity.py

6. Register a user: Navigate to http://localhost:7777 in your browser. The first user registered will be granted 超级管理员 (Super Administrator) privileges with full access.

# Clone and enter directory
$ git clone https://github.com/wuranxu/pity
$ cd pity

# Install dependencies
$ pip install -r requirements.txt

# Start the service
$ python pity.py

You can deploy the Pity platform using the provided docker-compose.yaml file. The setup includes three main services: a MySQL database, a Redis instance, and the Pity server itself, all connected via a shared network named pity_net.

Service Details

• MySQL (pity_mysql): Uses mysql:8.0.
  • Default Root Password: Pitytester666666
  • Default Database: pity
• Redis (pity_redis): Uses the standard redis image.
  • Default Password: 123456 (set via --requirepass)
  • Timezone: Asia/Shanghai
• Pity Server (pity_server): Built from the local context using ops/dockerfile.
  • Port Mapping: 7777:7777
  • Configuration: Mounts ./dev.env.ops to /pity/conf/dev.env inside the container.
  • Timezone: Asia/Shanghai

version: "3"
services:
  mysql:
    image: mysql:8.0
    container_name: pity_mysql
    restart: always
    environment:
      MYSQL_ROOT_PASSWORD: Pitytester666666
      MYSQL_DATABASE: pity
    networks:
      pity_net:
        aliases:
          - pity_mysql

  redis:
    image: redis
    container_name: pity_redis
    command: redis-server --requirepass 123456
    restart: always
    environment:
      - TZ=Asia/Shanghai
    networks:
      pity_net:
        aliases:
          - pity_redis

  pity:
    build:
      context: ../
      dockerfile: ops/dockerfile
      args:
        buildno: 1
    container_name: pity_server
    ports:
      - "7777:7777"
    restart: always
    environment:
      - TZ=Asia/Shanghai
    networks:
      pity_net:
        aliases:
        - pity_server
    volumes:
      - ./dev.env.ops:/pity/conf/dev.env

networks:
  pity_net:

The Pity application uses FastAPI startup events to initialize core services. When the application starts, it performs the following sequence:

1. Redis Initialization: Attempts to connect to Redis via RedisHelper.ping(). If Config.REDIS_ON is true and the connection fails, the application will raise an error and fail to start. If Config.REDIS_ON is false, it logs a warning and continues.
2. Scheduler Initialization: Sets up AsyncIOScheduler using SQLAlchemyJobStore for persistent job storage. The job store uses Config.SQLALCHEMY_DATABASE_URI.
3. Database Initialization: Runs create_table() asynchronously to ensure the database schema is ready.

Developers extending the startup logic should use the @pity.on_event('startup') decorator.

@pity.on_event('startup')
async def init_redis():
    # Logic for Redis connection
    ...

@pity.on_event('startup')
def init_scheduler():
    # Logic for ApScheduler
    ...

@pity.on_event('startup')
async def init_database():
    # Logic for table creation
    ...

Pity provides a WebSocket endpoint at /ws/{user_id} for real-time messaging and heartbeat monitoring.

Key behaviors:

• Connection Management: Uses ws_manage.connect(websocket, user_id) to track active connections.
• Unread Notifications: Upon connection, the server checks for unread messages via PityNotificationDao.list_messages. If unread messages exist, it pushes a WebSocketMessage.msg_count JSON payload to the client.
• Protocol/Commands: The server responds to specific text commands:
  • HELLO SERVER (case-insensitive): Returns hello {user_id}.
  • HEARTBEAT (case-insensitive): Returns the {user_id}.
• Disconnection: Automatically handles WebSocketDisconnect to clean up the connection in ws_manage.

Message Format Example (Unread Count): When unread messages are found, the server sends a JSON object structured by WebSocketMessage.msg_count.

@pity.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: int):
    # Implementation handles connection, unread message push, and command responses
    ...

The RpcClient class provides a mechanism to discover and connect to gRPC services registered in Etcd. Use get_instance(service) to retrieve an asynchronous client for a specific service. The method performs service discovery via Etcd, iterates through available addresses, and performs a HealthCheck() to ensure the service is in a working state before returning the instance. If a service exists but the requested method is missing (causing an AttributeError), it returns the instance as-is. If no healthy instances are found, it raises an Exception.

import asyncio
from app.utils.client import RpcClient

async def main():
    # Get an instance of the 'user' service
    service_instance = await RpcClient.get_instance("user")
    
    # Call a method on the service (e.g., loginV2)
    response = await service_instance.loginV2(dict(username="woody"))
    print(response)

asyncio.run(main())