Skip to content

Commit

Permalink
Dev javi/add unit tests for twitter services (#8)
Browse files Browse the repository at this point in the history
* Code refactor to implement unit testing

* Udpate comments.

* Update content.

* Content update.
  • Loading branch information
jke94 authored Feb 8, 2025
1 parent 7d3e372 commit 0ba4b68
Show file tree
Hide file tree
Showing 7 changed files with 230 additions and 115 deletions.
24 changes: 13 additions & 11 deletions Readme.md
Original file line number Diff line number Diff line change
@@ -1,27 +1,29 @@
# weather-station

[![On push master morning_daily_scheduler.py](https://github.com/jke94/weather-station/actions/workflows/master_on_push_morning_daily_scheduler.yml/badge.svg)](https://github.com/jke94/weather-station/actions/workflows/master_on_push_morning_daily_scheduler.yml)

[![Morning daily scheduler yesterday tweet in X (Twitter) at 09h UTC](https://github.com/jke94/weather-station/actions/workflows/morning_daily_scheduler_tweet_in_X.yml/badge.svg)](https://github.com/jke94/weather-station/actions/workflows/morning_daily_scheduler_tweet_in_X.yml)

[![Morning daily scheduler yesterday report 09h UTC](https://github.com/jke94/weather-station/actions/workflows/morning_daily_scheduler.yml/badge.svg)](https://github.com/jke94/weather-station/actions/workflows/morning_daily_scheduler.yml)

## Description

Weather station automatization reports using with Github Actions and X (also knowing as Twitter in the past) daily report publication process.

The information is obtained from [Weather Underground](https://www.wunderground.com/) platform. Launching HTTP request over the Web API to get information about the weather information sent by our weather station.

Finally it´s reported in X platform [
⛅Tiempo Castrocontrigo](https://x.com/Castro_tiempo)✌️
Finally it´s reported in X platform:

- X (Twitter) profile: [⛅Tiempo Castrocontrigo](https://x.com/Castro_tiempo)✌️

Here, a picture about the weather station 📸

<p align="center">
<img src="https://github.com/jke94/weather-station/blob/master/images/profile_photo_weather_station.jpg"
alt="Weather station photo"
style="max-width:450px; max-height:450px; height:auto;">
</p>

Github action status:

- [![On push master morning_daily_scheduler.py](https://github.com/jke94/weather-station/actions/workflows/master_on_push_morning_daily_scheduler.yml/badge.svg)](https://github.com/jke94/weather-station/actions/workflows/master_on_push_morning_daily_scheduler.yml)

- [![Morning daily scheduler yesterday tweet in X (Twitter) at 09h UTC](https://github.com/jke94/weather-station/actions/workflows/morning_daily_scheduler_tweet_in_X.yml/badge.svg)](https://github.com/jke94/weather-station/actions/workflows/morning_daily_scheduler_tweet_in_X.yml)

- [![Morning daily scheduler yesterday report 09h UTC](https://github.com/jke94/weather-station/actions/workflows/morning_daily_scheduler.yml/badge.svg)](https://github.com/jke94/weather-station/actions/workflows/morning_daily_scheduler.yml)


## How to run

0. Recommended create python virtual environment and install **requirements.txt** packages.
Expand Down
26 changes: 8 additions & 18 deletions src/weather-station/post_morning_daily_tweet_in_x.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,8 @@

from services.twitter_service import TwitterService
from services.twitter_service import build_tweet
from services.twitter_service import create_tweet

from services.twitter_create_post_service import create_tweet

def main(
weather_underground_station_id:str,
Expand Down Expand Up @@ -58,32 +59,21 @@ def main(
print(weather_day_summary_report.model_dump_json(indent=4))

# Call to Twitter (or also called X) service to build tweet content.
twitter_service = TwitterService(
build_tweet=build_tweet,
create_tweet=create_tweet
)

tweet_message = twitter_service.build_tweet(weather_day_summary_report)
twitter_service = TwitterService(create_tweet, build_tweet)

# Development: Show tweet info.
print('Tweet content --------------------------------------')
print(tweet_message)
print('--------------------------------------')
print(f'Tweet length: {len(tweet_message)}')

tweet_url = twitter_service.create_tweet(
tweet = twitter_service.post_weather_report(
api_key_for_x,
api_key_secret_for_x,
access_token_for_x,
access_secret_token_for_x,
tweet_message
weather_day_summary_report
)

if tweet_url == "NONE":
if tweet == "NONE":
print("Error in create tweet process!")
return -3
print(f'Created tweet: {tweet_url}')

print(f'Created tweet: {tweet}')

return 0

Expand Down
34 changes: 34 additions & 0 deletions src/weather-station/services/twitter_create_post_service.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
import tweepy

def create_tweet(
api_key,
api_secret,
access_token,
access_secret,
tweet_content
) -> str:

created_tweet_url = "NONE"

try:

# Create X (Twitter) client.
client = tweepy.Client(
consumer_key=api_key,
consumer_secret=api_secret,
access_token=access_token,
access_token_secret=access_secret
)

response = client.create_tweet(
text=tweet_content
)

created_tweet_url = f"https://twitter.com/user/status/{response.data['id']}"

return created_tweet_url

except Exception as error:

print({'error': f"Create tweet HTTP status code: {error}"})
return created_tweet_url
125 changes: 43 additions & 82 deletions src/weather-station/services/twitter_service.py
Original file line number Diff line number Diff line change
@@ -1,97 +1,58 @@
import tweepy
from typing import Callable, Optional
from typing import Callable

from model.report.weather_day_summary_report import WeatherDaySummaryReport

def trasnlate_to_spanish_uv_risk(uv_risk:str) -> str:

value = "_"

match uv_risk:
case "Low":
value = "Bajo"
case "Medium":
value = "Medio"
case "High":
value = "Alto"
case "Very high":
value = "Muy alto"
case "Extremely high":
value = "Extremo"
case "Very high":
value = "Muy alto"
case _:
value = "NONE"

return value
def translate_to_spanish_uv_risk(uv_risk: str) -> str:
translation = {
"Low": "Bajo",
"Medium": "Medio",
"High": "Alto",
"Very high": "Muy alto",
"Extremely high": "Extremo"
}
return translation.get(uv_risk, "NONE")

def build_tweet(weather_day_summary_report: WeatherDaySummaryReport) -> str:

date_str = weather_day_summary_report.Date

temperature_max = weather_day_summary_report.TemperatureHigh
temperature_low = weather_day_summary_report.TemperatureLow
temperature_avg = weather_day_summary_report.TemperatureAvg
precipitation_total = weather_day_summary_report.PrecipitationTotal
wind_speed_avg = weather_day_summary_report.WindSpeedAvg
wind_direction_avg = weather_day_summary_report.WindDirectionAvg
wind_gust_high = weather_day_summary_report.WindGustHigh
uv_high = weather_day_summary_report.UvIndexHigh
uv_risk = weather_day_summary_report.UvHighRisk
solar_radiation_high = weather_day_summary_report.SolarRadiationHigh

msg = (
return (
f"¡Buenos días Castrocontrigo!\n"
f"Resumen de ayer 📅 {date_str}:\n\n"
f"🌡️ Temp. (ºC): Max. {temperature_max} | Min. {temperature_low} | Media. {temperature_avg}\n"
f"💧 Lluvia: {precipitation_total} L/m²\n"
f"💨 Viento medio: {wind_speed_avg} km/h | Dir. media: {wind_direction_avg}\n"
f"🌀 Racha viento max.: {wind_gust_high} km/h\n"
f"☀️ Índice UV max.: {uv_high} ({trasnlate_to_spanish_uv_risk(uv_risk)})\n"
f"😎 Radiación solar max.: {solar_radiation_high} W/m²"
f"Resumen de ayer 📅 {weather_day_summary_report.Date}:\n\n"
f"🌡️ Temp. (ºC): Max. {weather_day_summary_report.TemperatureHigh} | "
f"Min. {weather_day_summary_report.TemperatureLow} | "
f"Media. {weather_day_summary_report.TemperatureAvg}\n"
f"💧 Lluvia: {weather_day_summary_report.PrecipitationTotal} L/m²\n"
f"💨 Viento medio: {weather_day_summary_report.WindSpeedAvg} km/h | "
f"Dir. media: {weather_day_summary_report.WindDirectionAvg}\n"
f"🌀 Racha viento max.: {weather_day_summary_report.WindGustHigh} km/h\n"
f"☀️ Índice UV max.: {weather_day_summary_report.UvIndexHigh} "
f"({translate_to_spanish_uv_risk(weather_day_summary_report.UvHighRisk)})\n"
f"😎 Radiación solar max.: {weather_day_summary_report.SolarRadiationHigh} W/m²"
)

return msg

def create_tweet(
api_key_for_x:str,
api_key_secret_for_x:str,
access_token_for_x:str,
access_secret_token_for_x:str,
tweet_content:str
) -> str:

created_tweet_url = "NONE"

try:

# Create X (Twitter) client.
client = tweepy.Client(
consumer_key=api_key_for_x,
consumer_secret=api_key_secret_for_x,
access_token=access_token_for_x,
access_token_secret=access_secret_token_for_x
)

response = client.create_tweet(
text=tweet_content
)

created_tweet_url = f"https://twitter.com/user/status/{response.data['id']}"

return created_tweet_url

except Exception as error:

print({'error': f"Create tweet HTTP status code: {error}"})
return created_tweet_url

class TwitterService:

def __init__(
self,
create_tweet: Callable[[str, str, str, str, str], str],
build_tweet: Callable[[WeatherDaySummaryReport], str]
):
):
self.build_tweet = build_tweet
self.create_tweet = create_tweet
self.create_tweet = create_tweet

def post_weather_report(
self,
api_key: str,
api_secret: str,
access_token: str,
access_secret: str,
weather_report: WeatherDaySummaryReport
) -> str:

tweet_content = self.build_tweet(weather_report)

return self.create_tweet(
api_key,
api_secret,
access_token,
access_secret,
tweet_content
)
4 changes: 2 additions & 2 deletions src/weather-station/test/run_unit_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,10 @@
import sys
import os

# Obtener la ruta absoluta del directorio actual (donde está run_unit_tests.py)
# Get the absolute path of the current directory (where run_unit_tests.py is)
CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))

# Agregar al sys.path el directorio 'weather-station' (dos niveles arriba)
# Add the 'weather-station' directory to the sys.path (two levels up)
PROJECT_ROOT = os.path.abspath(os.path.join(CURRENT_DIR, ".."))
sys.path.insert(0, PROJECT_ROOT)

Expand Down
94 changes: 94 additions & 0 deletions src/weather-station/test/unit_tests/test_twitter_service.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
import unittest
from unittest.mock import MagicMock
from model.report.weather_day_summary_report import WeatherDaySummaryReport
from services.twitter_service import TwitterService, build_tweet, translate_to_spanish_uv_risk

class TestTwitterService(unittest.TestCase):

def setUp(self):
"""Configure mocks"""
self.mock_create_tweet = MagicMock(return_value="https://twitter.com/user/status/123456789")

self.weather_report = WeatherDaySummaryReport(
Date="2024-02-03",
TemperatureHigh=14.4,
TemperatureAvg=7.7,
TemperatureLow=2.2,
DewPointHigh=6.4,
DewPointLow=-0.8,
DewPointAvg=2.7,
HumidityHigh=99.0,
HumidityLow=40.0,
HumidityAvg=72.7,
PrecipitationTotal=15.6,
PressureMax=1008.13,
PressureMin=1002.37,
WindSpeedHigh=19.8,
WindSpeedAvg=4.6,
WindGustHigh=25.4,
WindGustAvg=6.3,
WindDirectionAvg="WNW",
UvHighRisk="Medium",
UvIndexHigh=5.0,
SolarRadiationHigh=460.1
)

self.twitter_service = TwitterService(
create_tweet=self.mock_create_tweet,
build_tweet=build_tweet
)

def test_build_tweet_generates_correct_message(self):

# Arrange
expected_values = [
f"Resumen de ayer 📅 {self.weather_report.Date}",
f"🌡️ Temp. (ºC): Max. {self.weather_report.TemperatureHigh} | "
f"Min. {self.weather_report.TemperatureLow} | "
f"Media. {self.weather_report.TemperatureAvg}",
f"💧 Lluvia: {self.weather_report.PrecipitationTotal} L/m²",
f"💨 Viento medio: {self.weather_report.WindSpeedAvg} km/h | "
f"Dir. media: {self.weather_report.WindDirectionAvg}",
f"🌀 Racha viento max.: {self.weather_report.WindGustHigh} km/h",
f"☀️ Índice UV max.: {self.weather_report.UvIndexHigh} "
f"({translate_to_spanish_uv_risk(self.weather_report.UvHighRisk)})",
f"😎 Radiación solar max.: {self.weather_report.SolarRadiationHigh} W/m²"
]

# Act
tweet = build_tweet(self.weather_report)

# Assert
for value in expected_values:
with self.subTest(value=value):
self.assertIn(value, tweet)

def test_post_weather_report_calls_create_tweet_correctly(self):
"""Act: Call `post_weather_report`. Assert: Validate call to `create_tweet`."""

tweet_url = self.twitter_service.post_weather_report(
"API_KEY_FAKE",
"API_SECRET_FAKE",
"ACCESS_TOKEN_FAKE",
"ACCESS_SECRET_FAKE", self.weather_report
)

self.mock_create_tweet.assert_called_once_with(
"API_KEY_FAKE",
"API_SECRET_FAKE",
"ACCESS_TOKEN_FAKE",
"ACCESS_SECRET_FAKE",
build_tweet(self.weather_report)
)

self.assertEqual(tweet_url, "https://twitter.com/user/status/123456789")

def test_translate_to_spanish_uv_risk(self):
"""Act & Assert: Testing translation of different UV levels."""

self.assertEqual(translate_to_spanish_uv_risk("Low"), "Bajo")
self.assertEqual(translate_to_spanish_uv_risk("Medium"), "Medio")
self.assertEqual(translate_to_spanish_uv_risk("High"), "Alto")
self.assertEqual(translate_to_spanish_uv_risk("Very high"), "Muy alto")
self.assertEqual(translate_to_spanish_uv_risk("Extremely high"), "Extremo")
self.assertEqual(translate_to_spanish_uv_risk("Unknown"), "NONE")
Loading

0 comments on commit 0ba4b68

Please sign in to comment.