1

我最近编写了一个客户端测试程序来测试我的服务器。在测试中，我开始使用 orm 删除数据库中现有的对象（行）。然后我向服务器发送一系列命令以重新创建数据。但是，当我尝试在测试程序中查询“已删除”对象时，除非我使用另一个“名称”的会话进行查询，否则查询将不会返回任何对象。我会担心会话在 SqlAlchemy 中的工作方式，并且不确定这是否是一个错误。下面将进一步说明问题。

engine  = create_engine("mysql:....")
Session = sessionmaker(bind=engine)
session = Session()

for registration in session.query(Registration).filter_by(license_id = id).all():
    session.delete(registration)
    session.commit()

.... 向服务器发送命令的代码，这将在数据库的注册表中创建一些行 ....

x = session.query(Registration).filter_by(license_id = id).all()
print x

x 将是一个空列表。

我尝试在第二次查询之前添加以下内容并且都返回相同的结果

session.close()
session = Session()

或者

session.expunge_all()

或者

session.exipre_all()

在我执行以下操作之前，上述任何一项都不起作用

new_session = Session()
x = new_session.query(Registration).filter_by(license_id = id).all()

这一次，x 返回存储在数据库中的内容。

我有点困惑为什么我需要更改会话变量名称才能工作。所有与会话相关的文档，例如 session.expire、session.expunge，甚至获取另一个会话实例（同名但不同的实例，我已经检查过），都不起作用。

感谢是否有人可以提供帮助。谢谢。

完整代码

import socket
import blowfish
import base64
import string
import sys
import logging
import time
from datetime import date, timedelta

from sqlalchemy import create_engine, and_
from sqlalchemy import Column, Integer, String, DateTime, Date, Boolean, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.exc import NoResultFound
from sqlalchemy.orm.session import Session

from lrp_modal import User, Product, License, Registration

log = logging.getLogger('')
log.setLevel(logging.INFO)
log.addHandler(logging.StreamHandler())

def test(token, passwd, xmsg, expect_ok):
    serverHost = 'localhost'
    serverPort = 9120

    sockobj = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sockobj.connect((serverHost, serverPort))
    data = sockobj.recv(1024)
    log.debug('Client received: %s', repr(data))

    log.debug("Send: %s",  xmsg)
    sockobj.send(xmsg)

    x = ''
    while 1:
        data = sockobj.recv(1024)
        if not data:
            break
        x += data
    x = x.strip()
    log.debug("Receive: %s", x)
    if expect_ok and not x.startswith('<lrp'):
        print "Test fail"
        sys.exit(1)
    if not expect_ok and x.startswith('<lrp'):
        print "Test fail"
        sys.exit(1)

master   = 'X01X-0000-0000-'
liclist  = {
    'product60_tri': ['0001'],
    'product60_pro': ['0002'],
    'product50_ent': ['0003', '0004', '0005'],
    'product60_ent': ['0006', '0007', '0008'],
    }

machines = ['FFBCC89220EF', '00112233']

# Setup the licenses testing data
engine  = create_engine('mysql://root@127.0.0.1/padsystem')
Session = sessionmaker(bind=engine)
session = Session()

for tag, idlist in liclist.items():
    for id in idlist:
        product = session.query(Product).filter_by(product_tag = tag).one()
        try:
            license = session.query(License).filter_by(license_id = master + id).one()
        except NoResultFound:
            continue

        license.product_id = product.product_id
        license.act_rid  = ''
        license.act_type = ''
        license.act_hint = ''
        license.group_id = ''
        if id == '0003':
            license.lic_expire = date(2011, 1, 1)
            if id in ('0001', '0006'):
                license.lic_expire = date.today() + timedelta(10)
        if id in ('0004', '0005', '0007', '0008'):
            license.group_id = 'X'
        session.commit()

        for registration in session.query(Registration).filter_by(license_id = license.license_id).all():
            session.delete(registration)
        session.commit()

user    = session.query(User).get(1)

# Test to register a machine
log.info('Test register machine...')
data = """<lrp service="register">
    <registration>
      <product>Product50</product>
      <license>%s</license>
      <machine>%s</machine>
    </registration>
  </lrp>
  """
for id in ('0004', '0006', '0007'):
    lic_id = master + id
    test('LICENSEE %s\n' % user.user_name, user.password, data % (lic_id, machines[0]), True)
    test('LICENSEE %s\n' % user.user_name, user.password, data % (lic_id, machines[1]), True)

for id in ('0001', '0003'):
    lic_id = master + id
    test('LICENSEE %s\n' % user.user_name, user.password, data % (lic_id, machines[0]), False)

# Test to register a machine until register count is full.
lic_id = master + '0002'
for i in range(5):
    log.info("Register count test %s ", i)
    x = data % (lic_id, machines[1] + str(i))
    test('LICENSEE %s\n' % user.user_name, user.password, x, True)

log.info("Register exceed max...")
x = data % (lic_id, machines[0])
test('LICENSEE %s\n' % user.user_name, user.password, x, False)

# Test register to group of licenses
lic_id = master + '0008'
for i in range(8):
    log.info("Register group count test: %s ", i)
    x = data % (lic_id, machines[1] + str(i))
    test('LICENSEE %s\n' % user.user_name, user.password, x, True)

log.info("Register exceed group max...")
x = data % (lic_id, machines[0])
test('LICENSEE %s\n' % user.user_name, user.password, x, False)

# Test to unregister
data = """<lrp service="unregister">
    <registration id="%s">
      <product>Product50</product>
      <license>%s</license>
      <machine>%s</machine>
    </registration>
  </lrp>
  """

# session.expunge_all()  not work
# session.expire_all()   not work
# session.close(); del(session); session = Session()  not work
new_session = Session()  # This is needed, cannot use session!!
lic_id = master + '0004'
# Use original session instance will get null list.
for r in new_session.query(Registration).filter_by(license_id = lic_id).all():
    rid = r.rid
    mid = r.machine_id
    log.info("Unregister, wrong license...")
    x = data % (rid, master + '0005', mid)
    test('REGISTER %s\n' % rid, mid, x, False)
    log.info("Unregister...")
    x = data % (rid, lic_id, mid) 
    test('REGISTER %s\n' % rid, mid, x, True)
4

0 回答 0