test_tool_provider.py 1.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940
  1. import os
  2. from dotenv import load_dotenv
  3. import psycopg2
  4. load_dotenv()
  5. def test_tool_provider():
  6. print("Testing connection to tool_provider table...")
  7. try:
  8. conn = psycopg2.connect(
  9. host=os.getenv('KNOWHUB_DB'),
  10. port=int(os.getenv('KNOWHUB_PORT', 5432)),
  11. user=os.getenv('KNOWHUB_USER'),
  12. password=os.getenv('KNOWHUB_PASSWORD'),
  13. database=os.getenv('KNOWHUB_DB_NAME')
  14. )
  15. c = conn.cursor()
  16. # Check if tool_provider table exists by selecting 0 rows
  17. c.execute("SELECT * FROM tool_provider LIMIT 0")
  18. print("Success! Table tool_provider exists.")
  19. # Ensure it has the correct columns
  20. expected_cols = ['tool_id', 'provider_id']
  21. col_names = [desc[0] for desc in c.description]
  22. print(f"Columns in tool_provider: {col_names}")
  23. # Check if there's any data
  24. c.execute("SELECT * FROM tool_provider LIMIT 5")
  25. rows = c.fetchall()
  26. print(f"Sample data from tool_provider: {rows}")
  27. cursor = conn.cursor()
  28. c.close()
  29. conn.close()
  30. except Exception as e:
  31. print(f"Failed to access tool_provider table: {e}")
  32. if __name__ == "__main__":
  33. test_tool_provider()