()
| 521 | |
| 522 | #测试sql是否能执行 |
| 523 | def test_sql(): |
| 524 | |
| 525 | dst_dir = '/www/test/db/' |
| 526 | if os.path.exists(dst_dir): shutil.rmtree(dst_dir) |
| 527 | |
| 528 | os.makedirs(dst_dir, 755) |
| 529 | |
| 530 | all_tabs = get_database_json(False) |
| 531 | for nkey in all_tabs.keys(): |
| 532 | db_info = all_tabs[nkey] |
| 533 | |
| 534 | print_x('正在同步表到 {} 数据库：{}'.format(db_info['table'], db_info['db'])) |
| 535 | src_field = get_table_byjson(db_info['table'], db_info['db']) |
| 536 | dst_obj = get_db_obj('{}/{}'.format(dst_dir, db_info['db'])) |
| 537 | if not check_exists_table(dst_obj, db_info['table']): |
| 538 | res = dst_obj.execute(src_field['sql']) |
| 539 | if str(res).find("error") != -1: |
| 540 | print_x("创建表 {} 错误：{}".format(db_info['table'], res)) |
| 541 | break |
| 542 | |
| 543 | dst_field = get_table_info(dst_obj, db_info['table']) |
| 544 | |
| 545 | n_list = [] |
| 546 | for field in src_field['field']: |
| 547 | if not field in n_list: |
| 548 | n_list.append(field) |
| 549 | |
| 550 | for field in dst_field['field']: |
| 551 | if not field in n_list: |
| 552 | n_list.append(field) |
| 553 | |
| 554 | for field in n_list: |
| 555 | if not field in src_field['field']: |
| 556 | print('databases.json文件 {} - {} fields 数组里缺少字段：{}'.format(db_info['db'],db_info['table'],field)) |
| 557 | |
| 558 | if not field in dst_field['field']: |
| 559 | print('databases.json文件 {} - {} sql语句里缺少字段：{}'.format(db_info['db'],db_info['table'],field)) |
| 560 | |
| 561 | """ |
| 562 | @name 检查数据库字段 |
no test coverage detected