Dieser Text ist der erste einer Reihe von Texten über Tests in Datenverarbeitungsanwendungen, die ich hier und auf meinem persönlichen Blog veröffentlichen werde.
Als ich meinen beruflichen Übergang vom Softwareentwickler zum Dateningenieur vollzog, begann ich Gespräche mit Leuten im Datenbereich zu führen, die keinen Hintergrund in der Softwareentwicklung hatten. In diesen Gesprächen tauchte immer wieder die Frage auf: Wie schreibt man Tests?
Das Schreiben von Tests kann für diejenigen, die damit nicht vertraut sind, tatsächlich wie eine komplexe Aufgabe erscheinen, da es eine Änderung der Art und Weise erfordert, Code zu schreiben. Die Wahrheit ist, dass es kein Geheimnis ist, sondern eine Frage der Übung und Wiederholung. Mein Hauptziel in diesem Artikel ist es, Sie, die gerade erst anfangen, durch einen Prozess zu führen, der zeigt, wie wir Tests für Anwendungen erstellen können, die Daten verarbeiten und so Qualität und Zuverlässigkeit im Code sicherstellen.
Dieser Text ist Teil einer Reihe, die ich in den nächsten Wochen vorstellen werde und in der ich erläutere, wie man automatisierte Tests in Code für die Datentechnik schreibt. Im heutigen Artikel möchte ich ein wenig über Mocks sprechen. In mehreren Codeszenarien stellt eine Datenpipeline Verbindungen, API-Aufrufe, Integrationen mit Cloud-Diensten usw. her, was zu Verwirrung darüber führen kann, wie wir diese Anwendung testen können. Heute werden wir einige interessante Bibliotheken zum Schreiben von Tests erkunden, die sich auf die Verwendung von Mocks konzentrieren.
Mocks sind Scheinobjekte, die in Tests verwendet werden, um das Verhalten externer Abhängigkeiten oder Komponenten zu imitieren, die nicht im Fokus des Tests stehen. Sie ermöglichen es Ihnen, die zu testende Codeeinheit zu isolieren und so sicherzustellen, dass das Testen besser kontrollierbar und prädiktiv ist. Die Verwendung von Mocks ist eine gängige Praxis bei Unit-Tests und Integrationstests.
Und wir sollten Mocks verwenden, wenn:
In Datenpipelines können Sie mit Mocking Darstellungen externer Komponenten – wie einer Datenbank, eines Messaging-Dienstes oder einer API – erstellen, ohne von deren realen Infrastrukturen abhängig zu sein. Dies ist besonders nützlich in Datenverarbeitungsumgebungen, die mehrere Technologien integrieren, wie z. B. PySpark für die verteilte Verarbeitung, Kafka für Messaging sowie Cloud-Dienste wie AWS und GCP.
In diesen Szenarien, in denen wir über Datenpipelines verfügen, erleichtert Mocking die Ausführung isolierter und schneller Tests und minimiert so Kosten und Ausführungszeit. Es ermöglicht die genaue Überprüfung jedes Teils der Pipeline, ohne zeitweilige Ausfälle, die durch echte Verbindungen oder externe Infrastruktur verursacht werden, und mit der Gewissheit, dass jede Integration wie erwartet funktioniert.
In jeder Programmiersprache finden wir interne Module, die bereits zu implementierende Mock-Funktionen bereitstellen. In Python ist die native Bibliothek unittest.mock das Hauptwerkzeug zum Erstellen von Mocks, mit der Sie Objekte und Funktionen einfach und kontrolliert simulieren können. In Go wird der Mocking-Prozess häufig von externen Paketen wie Mocking unterstützt, da die Sprache keine native Mock-Bibliothek hat; Mockery eignet sich besonders zum Generieren von Mocks aus Schnittstellen, einer nativen Funktion von Go. In Java ist Mockito eine beliebte und leistungsstarke Bibliothek zum Erstellen von Mocks, die sich in JUnit integrieren lässt, um robuste Unit-Tests zu ermöglichen. Diese Bibliotheken bieten eine wesentliche Grundlage für das Testen isolierter Komponenten, insbesondere in Datenpipelines und verteilten Systemen, bei denen die Simulation externer Datenquellen und APIs von entscheidender Bedeutung ist.
Beginnen wir mit einem einfachen Beispiel, wie wir Mocks verwenden können. Angenommen, wir haben eine Funktion, die API-Aufrufe durchführt, und wir müssen Komponententests für diese Funktion schreiben:
def get_data_from_api(url):
import requests
response = requests.get(url)
if response.status_code == 200:
return response.json()
else:
return None
Um Testszenarien richtig anzugehen, müssen wir zunächst verstehen, welche Situationen abgedeckt werden sollten. Da unsere Funktion REST-Aufrufe durchführt, müssen Tests mindestens zwei Hauptszenarien berücksichtigen: eines, in dem die Anfrage erfolgreich ist, und ein anderes, in dem die Antwort nicht wie erwartet ausfällt. Wir könnten den Code mit einer echten URL ausführen, um das Verhalten zu beobachten, aber dieser Ansatz hat Nachteile, da wir keine Kontrolle über die verschiedenen Antworttypen hätten und den Test außerdem anfällig für Änderungen in der URL-Antwort oder deren eventuelle Nichtverfügbarkeit machen würden . Um diese Inkonsistenzen zu vermeiden, verwenden wir Mocks.
from unittest import mock
@mock.patch('requests.get')
def test_get_data_from_api_success(mock_get):
# Configura o mock para retornar uma resposta simulada
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {"key": "value"}
# Chama a função com o mock ativo
result = get_data_from_api("http://fakeurl.com")
# Verifica se o mock foi chamado corretamente e o resultado é o esperado
mock_get.assert_called_once_with("http://fakeurl.com")
self.assertEqual(result, {"key": "value"})
Mit der @mock.patch-Dekoration aus der Python-Unittest-Bibliothek können wir den request.get-Aufruf durch ein Mock ersetzen, ein „falsches Objekt“, das das Verhalten der get-Funktion im Testkontext simuliert und so die externe Abhängigkeit beseitigt .
Indem wir Werte für den Rückgabewert des Mocks definieren, können wir genau angeben, was das Objekt zurückgeben soll, wenn es in der von uns getesteten Funktion aufgerufen wird. Es ist wichtig, dass die return_value-Struktur der Struktur der realen Objekte entspricht, die wir ersetzen. Beispielsweise verfügt ein Antwortobjekt aus dem Anforderungsmodul über Attribute wie „status_code“ und Methoden wie „json()“. Um also eine Antwort der Funktion „requests.get“ zu simulieren, können wir diesen Attributen und Methoden direkt im Mock den erwarteten Wert zuweisen.
def get_data_from_api(url):
import requests
response = requests.get(url)
if response.status_code == 200:
return response.json()
else:
return None
In diesem speziellen Fall liegt der Schwerpunkt auf der Simulation der Anforderungsantwort, d. h. auf dem Testen des Verhaltens der Funktion mit unterschiedlichen erwarteten Ergebnissen, ohne von einer externen URL abhängig zu sein und ohne Auswirkungen auf unsere Testumgebung.
from unittest import mock
@mock.patch('requests.get')
def test_get_data_from_api_success(mock_get):
# Configura o mock para retornar uma resposta simulada
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {"key": "value"}
# Chama a função com o mock ativo
result = get_data_from_api("http://fakeurl.com")
# Verifica se o mock foi chamado corretamente e o resultado é o esperado
mock_get.assert_called_once_with("http://fakeurl.com")
self.assertEqual(result, {"key": "value"})
Durch die Simulation von API-Fehlerreaktionen in Tests können wir über die Grundlagen hinausgehen und das Anwendungsverhalten anhand verschiedener Arten von HTTP-Statuscodes wie 404, 401, 500 und 503 prüfen. Dies bietet eine breitere Abdeckung und stellt sicher, dass die Anwendung angemessen reagiert Ich verstehe, wie sich diese Variationen im Aufruf bei jeder Art von Fehler auf unsere Anwendungs-/Datenverarbeitung auswirken können. Bei POST-Methodenaufrufen können wir eine zusätzliche Validierungsebene hinzufügen, die nicht nur den Statuscode und die Grundfunktion des Aufrufs überprüft, sondern auch das Schema der gesendeten und empfangenen Antwort und so sicherstellt, dass die zurückgegebenen Daten dem entsprechen erwartetes Format. Dieser detailliertere Testansatz hilft, zukünftige Probleme zu vermeiden, indem er sicherstellt, dass die Anwendung auf die Bewältigung einer Vielzahl von Fehlerszenarien vorbereitet ist und dass die empfangenen Daten immer mit dem übereinstimmen, was entworfen wurde.
Da wir nun einen einfachen Fall der Verwendung von Mocks in reinem Python-Code gesehen haben, erweitern wir unsere Fälle auf einen Codeausschnitt, der Pyspark verwendet.
Um PySpark-Funktionen zu testen, insbesondere DataFrame-Operationen wie Filter, GroupBy und Join, ist die Verwendung von Mocks ein effektiver Ansatz, der die Ausführung von echtem Spark überflüssig macht, die Testzeit verkürzt und die Entwicklungsumgebung vereinfacht. Mit der Unittest.mock-Bibliothek von Python können Sie das Verhalten dieser Methoden simulieren und so den Codefluss und die Logik ohne Abhängigkeit von der Spark-Infrastruktur überprüfen.
Sehen wir uns die folgende Funktion an, in der wir eine Transformation haben, die Filter-, GroupBy- und Join-Vorgänge für Datenrahmen in Spark ausführt.
def get_data_from_api(url):
import requests
response = requests.get(url)
if response.status_code == 200:
return response.json()
else:
return None
Um einen PySpark-Test durchzuführen, müssen wir die Spark-Konfiguration lokal durchführen. Diese Konfiguration erfolgt in der setUpClass-Methode, die eine Instanz von Spark erstellt, die in allen Tests der Klasse verwendet wird. Dadurch können wir PySpark isoliert ausführen und so echte Transformationsvorgänge durchführen, ohne auf einen vollständigen Cluster angewiesen zu sein. Nach Abschluss des Tests ist die Methode „tearDownClass“ dafür verantwortlich, die Spark-Sitzung zu beenden und sicherzustellen, dass alle Ressourcen ordnungsgemäß freigegeben werden und die Testumgebung sauber ist.
from unittest import mock
@mock.patch('requests.get')
def test_get_data_from_api_success(mock_get):
# Configura o mock para retornar uma resposta simulada
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {"key": "value"}
# Chama a função com o mock ativo
result = get_data_from_api("http://fakeurl.com")
# Verifica se o mock foi chamado corretamente e o resultado é o esperado
mock_get.assert_called_once_with("http://fakeurl.com")
self.assertEqual(result, {"key": "value"})
Im test_transform_data-Test erstellen wir zunächst Beispiel-DataFrames für df und df_other, die die Daten enthalten, die in den Transformationen verwendet werden. Anschließend führen wir die Funktion „transform_data“ aus, ohne Mocks anzuwenden, sodass die Filter-, GroupBy- und Join-Vorgänge tatsächlich ausgeführt werden und zu einem neuen DataFrame führen. Nach der Ausführung verwenden wir die Methode „collect()“, um die Daten aus dem resultierenden DataFrame zu extrahieren, was es uns ermöglicht, diese Daten mit den erwarteten Werten zu vergleichen und so die durchgeführte Transformation auf reale und genaue Weise zu validieren.
Aber wir können auch Szenarien haben, in denen wir das Ergebnis einer dieser Pyspark-Funktionen testen möchten. Es ist notwendig, einen anderen Teil des Codes zu verspotten, der zur Ausführungszeit möglicherweise einen Engpass darstellt und kein Risiko für unseren Prozess darstellt. Daher können wir die Technik des Verspottens einer Funktion/eines Moduls verwenden, wie wir im vorherigen Beispiel mithilfe von Anfragen gesehen haben.
response.status_code = mock_get.return_value.status_code response.json() = mock_get.return_value.json.return_value
Der Mock-Test für eine bestimmte Operation wurde in der Methode test_transform_data_with_mocked_join durchgeführt, wobei wir einen Mock speziell für die Filtermethode angewendet haben. Dieser Mock ersetzt das Ergebnis des Join-Vorgangs durch einen simulierten DataFrame, sodass frühere Vorgänge wie „groupBy“ und „join“ real ausgeführt werden können. Der Test vergleicht dann den resultierenden DataFrame mit dem erwarteten Wert und stellt so sicher, dass der Join-Mock korrekt verwendet wurde, ohne die anderen durchgeführten Transformationen zu beeinträchtigen.
Dieser hybride Ansatz bringt mehrere Vorteile mit sich. Indem wir sicherstellen, dass tatsächliche PySpark-Vorgänge wie „join“ und „groupBy“ beibehalten werden, können wir die Logik von Transformationen validieren, ohne die Flexibilität zu verlieren, bestimmte Vorgänge wie „Filter“ durch Mocks zu ersetzen. Dies führt zu robusteren und schnelleren Tests, wodurch die Notwendigkeit eines vollständigen Spark-Clusters entfällt, was die laufende Codeentwicklung und -validierung erleichtert.
Es ist wichtig hervorzuheben, dass diese Strategie mit Vorsicht und nur in Szenarien angewendet werden sollte, in denen keine Verzerrung der Ergebnisse entsteht. Der Zweck der Prüfung besteht darin, sicherzustellen, dass die Verarbeitung korrekt erfolgt; Wir sollten nicht einfach Werte zuweisen, ohne die Funktion tatsächlich zu testen. Obwohl es zulässig ist, Abschnitte zu verspotten, von denen wir garantieren können, dass sie den Unit-Testprozess nicht beeinträchtigen, ist es wichtig, sich daran zu erinnern, dass die Funktion ausgeführt werden muss, um ihr tatsächliches Verhalten zu überprüfen.
Daher ist der hybride Ansatz viel sinnvoller, wenn dieser Funktion weitere Verarbeitungsarten hinzugefügt werden. Diese Strategie ermöglicht eine effektive Kombination aus realen und simulierten Vorgängen und sorgt so für robustere und zuverlässigere Tests
Mocks sind wertvolle Verbündete bei der Erstellung effektiver Tests, insbesondere wenn es um die Arbeit mit PySpark und anderen Cloud-Diensten geht. Die Implementierung, die wir mit Unittest in Python untersucht haben, hat uns nicht nur dabei geholfen, Vorgänge zu simulieren, sondern auch die Integrität unserer Daten und Prozesse aufrechtzuerhalten. Dank der Flexibilität, die Mocks bieten, können wir unsere Pipelines testen, ohne befürchten zu müssen, dass es in Produktionsumgebungen zu Schäden kommt. Also, bereit für die nächste Herausforderung? In unserem nächsten Text tauchen wir in die Welt der Integrationen mit AWS- und GCP-Diensten ein und zeigen, wie Sie diese Aufrufe verspotten und sicherstellen, dass Ihre Pipelines perfekt funktionieren. Bis zum nächsten Mal!
Das obige ist der detaillierte Inhalt vonMocks, was sind sie?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
![[Web-Frontend] Node.js-Schnellstart](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
